Merge pull request #541 from arabcoders/dev
Some checks failed
native-build / build (amd64, ubuntu-latest) (push) Has been cancelled
native-build / build (amd64, windows-latest) (push) Has been cancelled
native-build / build (arm64, macos-latest) (push) Has been cancelled
native-build / build (arm64, ubuntu-latest) (push) Has been cancelled
native-build / build (arm64, windows-latest) (push) Has been cancelled

Fix: preset export was exporting cookies due to inverted check.
This commit is contained in:
Abdulmohsen 2026-01-11 21:36:42 +03:00 committed by GitHub
commit 3b49678272
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
30 changed files with 1218 additions and 356 deletions

View file

@ -160,6 +160,7 @@
"rejecttitle", "rejecttitle",
"remux", "remux",
"reqs", "reqs",
"restrictfilenames",
"RPAREN", "RPAREN",
"rtime", "rtime",
"rvfc", "rvfc",

25
API.md
View file

@ -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 ### GET /api/dev/loop
**Purpose**: Development-only. Show event loop details and running tasks. **Purpose**: Development-only. Show event loop details and running tasks.

1
FAQ.md
View file

@ -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_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_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_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] > [!NOTE]
> To raise the maximum workers for specific extractor, you need to add a ENV variable that follows the pattern `YTP_MAX_WORKERS_FOR_<EXTRACTOR_NAME>`. > To raise the maximum workers for specific extractor, you need to add a ENV variable that follows the pattern `YTP_MAX_WORKERS_FOR_<EXTRACTOR_NAME>`.

View file

@ -306,7 +306,7 @@ class Download:
raise ValueError(msg) # noqa: TRY301 raise ValueError(msg) # noqa: TRY301
self.logger.info( 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: if self.debug:
@ -336,7 +336,13 @@ class Download:
self.logger.debug(f"Downloading '{self.info.url}' using pre-info.") self.logger.debug(f"Downloading '{self.info.url}' using pre-info.")
_dct: dict = self.info_dict.copy() _dct: dict = self.info_dict.copy()
if isinstance(self.info.extras, dict) and len(self.info.extras) > 0: 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) cls.process_ie_result(ie_result=_dct, download=True)
ret: int = cls._download_retcode ret: int = cls._download_retcode

View file

@ -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

View file

@ -96,29 +96,34 @@ class FileLogFormatter(logging.Formatter):
return datetime.fromtimestamp(record.created).astimezone().isoformat(timespec="milliseconds") 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. Get a static YTDLP instance for info extraction.
Args: Args:
reload (bool): If True, forces re-creation of the instance. params (dict|None): YTDLP parameters.
Returns: Returns:
YTDLP: A static YTDLP instance. YTDLP: A static YTDLP instance.
""" """
global YTDLP_INFO_CLS # noqa: PLW0603 global YTDLP_INFO_CLS # noqa: PLW0603
if YTDLP_INFO_CLS is None or reload:
YTDLP_INFO_CLS = YTDLP( default_params: dict[str, Any] = {
params={ "color": "no_color",
"color": "no_color", "extract_flat": True,
"extract_flat": True, "skip_download": True,
"skip_download": True, "ignoreerrors": True,
"ignoreerrors": True, "ignore_no_formats_error": True,
"ignore_no_formats_error": True, "quiet": 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 return YTDLP_INFO_CLS
@ -1438,7 +1443,7 @@ def get_archive_id(url: str) -> dict[str, str | None]:
"archive_id": None, "archive_id": None,
} }
for key, _ie in get_static_ytdlp()._ies.items(): for key, _ie in get_ytdlp()._ies.items():
try: try:
if not _ie.suitable(url): if not _ie.suitable(url):
continue continue
@ -1937,16 +1942,17 @@ def get_extras(entry: dict, kind: str = "video") -> dict:
return extras 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. Parse yt-dlp output template with given info_dict.
Args: Args:
output_template (str): The output template string. output_template (str): The output template string.
info_dict (dict): The info dictionary from yt-dlp. info_dict (dict): The info dictionary from yt-dlp.
params (dict|None): Additional parameters for yt-dlp.
Returns: Returns:
str: The parsed output string. 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)

View file

@ -229,6 +229,12 @@ class Config(metaclass=Singleton):
static_ui_path: str = "" static_ui_path: str = ""
"The path to the static UI files." "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 = ( _manual_vars: tuple = (
"temp_path", "temp_path",
"config_path", "config_path",
@ -245,6 +251,7 @@ class Config(metaclass=Singleton):
"app_commit_sha", "app_commit_sha",
"app_build_date", "app_build_date",
"app_branch", "app_branch",
"new_version",
) )
"The variables that are immutable." "The variables that are immutable."
@ -284,6 +291,7 @@ class Config(metaclass=Singleton):
"allow_internal_urls", "allow_internal_urls",
"simple_mode", "simple_mode",
"ignore_archived_items", "ignore_archived_items",
"check_for_updates",
) )
"The variables that are booleans." "The variables that are booleans."
@ -314,6 +322,8 @@ class Config(metaclass=Singleton):
"app_build_date", "app_build_date",
"app_branch", "app_branch",
"default_pagination", "default_pagination",
"check_for_updates",
"new_version",
) )
"The variables that are relevant to the frontend." "The variables that are relevant to the frontend."

View file

@ -29,6 +29,7 @@ from app.library.Services import Services
from app.library.sqlite_store import SqliteStore from app.library.sqlite_store import SqliteStore
from app.library.TaskDefinitions import TaskDefinitions from app.library.TaskDefinitions import TaskDefinitions
from app.library.Tasks import Tasks from app.library.Tasks import Tasks
from app.library.UpdateChecker import UpdateChecker
LOG = logging.getLogger("app") LOG = logging.getLogger("app")
MIME = magic.Magic(mime=True) MIME = magic.Magic(mime=True)
@ -120,6 +121,7 @@ class Main:
DLFields.get_instance().attach(self._app) DLFields.get_instance().attach(self._app)
TaskDefinitions.get_instance().attach(self._app) TaskDefinitions.get_instance().attach(self._app)
DownloadQueue.get_instance().attach(self._app) DownloadQueue.get_instance().attach(self._app)
UpdateChecker.get_instance().attach(self._app)
EventBus.get_instance().emit( EventBus.get_instance().emit(
Events.LOADED, Events.LOADED,

View file

@ -11,6 +11,7 @@ from app.library.DownloadQueue import DownloadQueue
from app.library.encoder import Encoder from app.library.encoder import Encoder
from app.library.Events import EventBus, Events from app.library.Events import EventBus, Events
from app.library.router import route from app.library.router import route
from app.library.UpdateChecker import UpdateChecker
LOG: logging.Logger = logging.getLogger(__name__) 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, status=web.HTTPOk.status_code,
dumps=encoder.encode, 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,
)

View file

@ -2,6 +2,7 @@ import asyncio
import functools import functools
import logging import logging
import uuid import uuid
from pathlib import Path
from typing import TYPE_CHECKING, Any from typing import TYPE_CHECKING, Any
from aiohttp import web from aiohttp import web
@ -266,14 +267,16 @@ async def task_metadata(request: Request, config: Config, encoder: Encoder) -> R
if not task.folder: if not task.folder:
try: try:
outtmpl = parse_outtmpl( ytdlp_opts: dict = task.get_ytdlp_opts().get_all()
output_template=task.get_ytdlp_opts().get_all().get("outtmpl", {}).get("default", "{title} [{id}]"), outtmpl: str = parse_outtmpl(
output_template=ytdlp_opts.get("outtmpl", {}).get("default", "{title} [{id}]"),
info_dict=metadata, info_dict=metadata,
params=ytdlp_opts,
) )
if outtmpl: if outtmpl:
_path = save_path / outtmpl _path: Path = save_path / outtmpl
if not _path.is_dir(): 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)) (save_path, _) = get_file(config.download_path, _path.relative_to(config.download_path))
if not str(save_path or "").startswith(str(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 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 yt_dlp.utils import sanitize_filename
from app.yt_dlp_plugins.postprocessor.nfo_maker import NFOMakerPP from app.yt_dlp_plugins.postprocessor.nfo_maker import NFOMakerPP

View file

@ -30,16 +30,14 @@ class TestCondition:
"""Test creating a condition with default values.""" """Test creating a condition with default values."""
condition = Condition(name="test", filter="duration > 60") condition = Condition(name="test", filter="duration > 60")
# Check that ID is generated assert condition.id, "Check that ID is generated"
assert condition.id
assert isinstance(condition.id, str) assert isinstance(condition.id, str)
# Check required fields # Check required fields
assert condition.name == "test" assert condition.name == "test"
assert condition.filter == "duration > 60" assert condition.filter == "duration > 60"
# Check defaults assert condition.cli == "", "Check defaults"
assert condition.cli == ""
assert condition.extras == {} assert condition.extras == {}
assert condition.enabled is True assert condition.enabled is True
@ -235,8 +233,7 @@ class TestConditions:
assert result is conditions assert result is conditions
assert len(conditions._items) == 2 assert len(conditions._items) == 2
# Check first condition assert conditions._items[0].name == "short_videos", "Check first condition"
assert conditions._items[0].name == "short_videos"
assert conditions._items[0].filter == "duration < 300" assert conditions._items[0].filter == "duration < 300"
assert conditions._items[0].cli == "--format worst" assert conditions._items[0].cli == "--format worst"
assert conditions._items[0].extras == {"category": "short"} assert conditions._items[0].extras == {"category": "short"}
@ -244,8 +241,7 @@ class TestConditions:
assert conditions._items[0].priority == 0 assert conditions._items[0].priority == 0
assert conditions._items[0].description == "Download short videos" assert conditions._items[0].description == "Download short videos"
# Check second condition assert conditions._items[1].name == "music_videos", "Check second condition"
assert conditions._items[1].name == "music_videos"
assert conditions._items[1].filter == "title ~= 'music'" assert conditions._items[1].filter == "title ~= 'music'"
assert conditions._items[1].enabled is False assert conditions._items[1].enabled is False
assert conditions._items[1].priority == 5 assert conditions._items[1].priority == 5
@ -265,8 +261,7 @@ class TestConditions:
conditions = Conditions(file=file_path) conditions = Conditions(file=file_path)
conditions.load() conditions.load()
# Should have generated ID assert len(conditions._items) == 1, "Should have generated ID"
assert len(conditions._items) == 1
assert conditions._items[0].id assert conditions._items[0].id
assert conditions._items[0].name == "no_id_test" assert conditions._items[0].name == "no_id_test"
@ -286,8 +281,7 @@ class TestConditions:
conditions = Conditions(file=file_path) conditions = Conditions(file=file_path)
conditions.load() conditions.load()
# Should have generated empty extras assert len(conditions._items) == 1, "Should have generated empty extras"
assert len(conditions._items) == 1
assert conditions._items[0].extras == {} assert conditions._items[0].extras == {}
# Should call save due to changes # Should call save due to changes
@ -306,8 +300,7 @@ class TestConditions:
conditions = Conditions(file=file_path) conditions = Conditions(file=file_path)
conditions.load() conditions.load()
# Should have generated enabled=True assert len(conditions._items) == 1, "Should have generated enabled=True"
assert len(conditions._items) == 1
assert conditions._items[0].enabled is True assert conditions._items[0].enabled is True
# Should call save due to changes # Should call save due to changes
@ -326,8 +319,7 @@ class TestConditions:
conditions = Conditions(file=file_path) conditions = Conditions(file=file_path)
conditions.load() conditions.load()
# Should have generated priority=0 assert len(conditions._items) == 1, "Should have generated priority=0"
assert len(conditions._items) == 1
assert conditions._items[0].priority == 0 assert conditions._items[0].priority == 0
# Should call save due to changes # Should call save due to changes
@ -348,8 +340,7 @@ class TestConditions:
conditions = Conditions(file=file_path) conditions = Conditions(file=file_path)
conditions.load() conditions.load()
# Should have generated description='' assert len(conditions._items) == 1, "Should have generated description=''"
assert len(conditions._items) == 1
assert conditions._items[0].description == "" assert conditions._items[0].description == ""
# Should call save due to changes # Should call save due to changes
@ -383,8 +374,7 @@ class TestConditions:
conditions = Conditions(file=file_path) conditions = Conditions(file=file_path)
result = conditions.load() result = conditions.load()
# Should load only valid conditions assert result is conditions, "Should load only valid conditions"
assert result is conditions
assert len(conditions._items) == 1 assert len(conditions._items) == 1
assert conditions._items[0].name == "valid" assert conditions._items[0].name == "valid"
@ -731,14 +721,11 @@ class TestConditions:
test_condition = Condition(name="test", filter="duration > 60") test_condition = Condition(name="test", filter="duration > 60")
conditions._items = [test_condition] conditions._items = [test_condition]
# Test with None assert conditions.match(None) is None, "Test with None"
assert conditions.match(None) is None
# Test with empty dict assert conditions.match({}) is None, "Test with empty dict"
assert conditions.match({}) is None
# Test with non-dict assert conditions.match("not a dict") is None, "Test with non-dict"
assert conditions.match("not a dict") is None
@patch("app.library.conditions.match_str") @patch("app.library.conditions.match_str")
def test_match_filter_evaluation_error(self, mock_match_str): def test_match_filter_evaluation_error(self, mock_match_str):
@ -774,8 +761,7 @@ class TestConditions:
info_dict = {"duration": 150} info_dict = {"duration": 150}
result = conditions.match(info_dict) result = conditions.match(info_dict)
# Should skip disabled condition and match enabled one assert result is enabled_condition, "Should skip disabled condition and match enabled one"
assert result is enabled_condition
# Should only call match_str once for enabled condition # Should only call match_str once for enabled condition
mock_match_str.assert_called_once_with("duration > 120", info_dict) mock_match_str.assert_called_once_with("duration > 120", info_dict)
@ -800,8 +786,7 @@ class TestConditions:
info_dict = {"duration": 120} info_dict = {"duration": 120}
result = conditions.match(info_dict) result = conditions.match(info_dict)
# Should match high_priority first (priority=10) assert result is high_priority, "Should match high_priority first (priority=10)"
assert result is high_priority
# Should only call match_str once for highest priority condition # Should only call match_str once for highest priority condition
mock_match_str.assert_called_once_with("duration > 60", info_dict) mock_match_str.assert_called_once_with("duration > 60", info_dict)
@ -891,8 +876,7 @@ class TestConditions:
info_dict = {"duration": 120} info_dict = {"duration": 120}
result = conditions.single_match("disabled_single", info_dict) result = conditions.single_match("disabled_single", info_dict)
# Should return None because condition is disabled assert result is None, "Should return None because condition is disabled"
assert result is None
def test_single_match_invalid_inputs(self): def test_single_match_invalid_inputs(self):
"""Test single matching with invalid inputs.""" """Test single matching with invalid inputs."""
@ -900,14 +884,10 @@ class TestConditions:
file_path = Path(temp_dir) / "invalid_single_test.json" file_path = Path(temp_dir) / "invalid_single_test.json"
conditions = Conditions(file=file_path) conditions = Conditions(file=file_path)
# Test with empty conditions assert conditions.single_match("test", {"duration": 120}) is None, "Test with empty conditions"
assert conditions.single_match("test", {"duration": 120}) is None
# Test with None info assert conditions.single_match("test", None) is None, "Test with None info"
assert conditions.single_match("test", None) is None
# Test with empty info dict assert conditions.single_match("test", {}) is None, "Test with empty info dict"
assert conditions.single_match("test", {}) is None
# Test with non-dict info assert conditions.single_match("test", "not a dict") is None, "Test with non-dict info"
assert conditions.single_match("test", "not a dict") is None

View file

@ -523,8 +523,7 @@ class TestDataStore:
assert "id1" in ids assert "id1" in ids
assert "id2" in ids assert "id2" in ids
# Verify order is maintained (OrderedDict) assert result[0][0] == "id1", "Verify order is maintained (OrderedDict)"
assert result[0][0] == "id1"
assert result[1][0] == "id2" assert result[1][0] == "id2"
await db.close() await db.close()

View file

@ -71,8 +71,7 @@ class TestDLField:
"""Test creating a DLField with default values.""" """Test creating a DLField with default values."""
field = DLField(name="test_field", description="Test description", field="--test-option") field = DLField(name="test_field", description="Test description", field="--test-option")
# Check that ID is generated assert field.id, "Check that ID is generated"
assert field.id
assert isinstance(field.id, str) assert isinstance(field.id, str)
# Check required fields # Check required fields
@ -80,8 +79,7 @@ class TestDLField:
assert field.description == "Test description" assert field.description == "Test description"
assert field.field == "--test-option" assert field.field == "--test-option"
# Check defaults assert field.kind == FieldType.TEXT, "Check defaults"
assert field.kind == FieldType.TEXT
assert field.icon == "" assert field.icon == ""
assert field.order == 0 assert field.order == 0
assert field.value == "" assert field.value == ""
@ -272,8 +270,7 @@ class TestDLFields:
fields = DLFields(file=str(temp_file)) fields = DLFields(file=str(temp_file))
fields.load() fields.load()
# Should handle error gracefully and return empty list assert fields.get_all() == [], "Should handle error gracefully and return empty list"
assert fields.get_all() == []
@patch("app.library.dl_fields.Config") @patch("app.library.dl_fields.Config")
def test_dl_fields_load_missing_id_auto_generation(self, mock_config, temp_file): def test_dl_fields_load_missing_id_auto_generation(self, mock_config, temp_file):
@ -490,8 +487,7 @@ class TestDLFields:
fields.save(test_fields) fields.save(test_fields)
# Verify file was written assert temp_file.exists(), "Verify file was written"
assert temp_file.exists()
# Verify content # Verify content
saved_data = json.loads(temp_file.read_text()) saved_data = json.loads(temp_file.read_text())

View file

@ -55,8 +55,7 @@ class TestNestedLogger:
assert levels.count(logging.INFO) == 1 assert levels.count(logging.INFO) == 1
msgs = [r.getMessage() for r in cap.records] msgs = [r.getMessage() for r in cap.records]
assert "[debug]" not in msgs[0] assert "[debug]" not in msgs[0]
# [download] prefix is not stripped by NestedLogger assert msgs[1] == "[download] progress", "[download] prefix is not stripped by NestedLogger"
assert msgs[1] == "[download] progress"
assert msgs[2] == "info message" assert msgs[2] == "info message"
@ -112,8 +111,7 @@ class TestDownloadHooks:
ev = q.items[0] ev = q.items[0]
assert ev["id"] == d.id assert ev["id"] == d.id
assert ev["action"] == "progress" assert ev["action"] == "progress"
# ensure only whitelisted keys included assert "other" not in ev, "ensure only whitelisted keys included"
assert "other" not in ev
for k in ( for k in (
"tmpfilename", "tmpfilename",
"filename", "filename",

View file

@ -55,8 +55,9 @@ class TestFFProbe:
"""Test that ffprobe results are cached with enhanced async timed_lru_cache.""" """Test that ffprobe results are cached with enhanced async timed_lru_cache."""
from app.library.ffprobe import ffprobe from app.library.ffprobe import ffprobe
# Test that the function has been decorated with caching assert hasattr(ffprobe, "cache_clear"), (
assert hasattr(ffprobe, "cache_clear"), "ffprobe should have cache_clear method from timed_lru_cache" "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" assert hasattr(ffprobe, "cache_info"), "ffprobe should have cache_info method from timed_lru_cache"
# Clear cache to start fresh # Clear cache to start fresh
@ -91,12 +92,13 @@ class TestFFProbe:
assert result2 is not None assert result2 is not None
assert isinstance(result2.metadata, dict) assert isinstance(result2.metadata, dict)
# The subprocess should not be called again for the actual ffprobe execution assert call_count == first_call_count, (
# (it may be called for the -h check, but the main execution should be cached) "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"
assert call_count == first_call_count, "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 @pytest.mark.asyncio
async def test_ffprobe_with_path_object(self): async def test_ffprobe_with_path_object(self):
@ -124,8 +126,7 @@ class TestFFProbe:
result = FFProbeResult() result = FFProbeResult()
# Test empty result assert result.video == [], "Test empty result"
assert result.video == []
assert result.audio == [] assert result.audio == []
assert result.subtitle == [] assert result.subtitle == []
assert result.attachment == [] assert result.attachment == []

View file

@ -32,8 +32,7 @@ class TestItemFormatAndBasics:
item = Item.format(data) item = Item.format(data)
assert isinstance(item, Item) assert isinstance(item, Item)
# URL normalized to full YouTube URL assert item.url.startswith("https://www.youtube.com/watch?v="), "URL normalized to full YouTube URL"
assert item.url.startswith("https://www.youtube.com/watch?v=")
assert item.preset == "custom" assert item.preset == "custom"
assert item.folder == "media" assert item.folder == "media"
assert item.cookies == "abc" assert item.cookies == "abc"
@ -146,8 +145,7 @@ class TestItemDTO:
def test_archive_add_and_delete_paths(self): def test_archive_add_and_delete_paths(self):
dto = ItemDTO(id="id", title="t", url="u", folder="f") dto = ItemDTO(id="id", title="t", url="u", folder="f")
# Precondition not met yet assert dto.archive_add() is False, "Precondition not met yet"
assert dto.archive_add() is False
# Set up to allow add # Set up to allow add
dto.archive_id = "arch" dto.archive_id = "arch"

View file

@ -216,13 +216,11 @@ class TestNotificationEvents:
def test_is_valid(self): def test_is_valid(self):
"""Test is_valid static method.""" """Test is_valid static method."""
# Valid events assert NotificationEvents.is_valid("test"), "Valid events"
assert NotificationEvents.is_valid("test")
assert NotificationEvents.is_valid("item_added") assert NotificationEvents.is_valid("item_added")
assert NotificationEvents.is_valid("item_completed") assert NotificationEvents.is_valid("item_completed")
# Invalid events assert not NotificationEvents.is_valid("invalid_event"), "Invalid events"
assert not NotificationEvents.is_valid("invalid_event")
assert not NotificationEvents.is_valid("") assert not NotificationEvents.is_valid("")
assert not NotificationEvents.is_valid(None) assert not NotificationEvents.is_valid(None)
@ -417,8 +415,7 @@ class TestNotification:
# Verify save was called due to schema update # Verify save was called due to schema update
mock_save.assert_called_once() mock_save.assert_called_once()
# Verify the target has enabled=True by default assert len(notification._targets) == 1, "Verify the target has enabled=True by default"
assert len(notification._targets) == 1
assert notification._targets[0].enabled is True assert notification._targets[0].enabled is True
# Clean up # Clean up
@ -744,8 +741,7 @@ class TestNotification:
result = await notification.send(event) result = await notification.send(event)
# Only enabled target should be called assert len(result) == 1, "Only enabled target should be called"
assert len(result) == 1
assert result[0]["status"] == 200 assert result[0]["status"] == 200
mock_client.request.assert_called_once() mock_client.request.assert_called_once()
@ -783,8 +779,7 @@ class TestNotification:
result = await notification.send(event) result = await notification.send(event)
# Should return empty dict from _apprise method assert len(result) == 1, "Should return empty dict from _apprise method"
assert len(result) == 1
assert result[0] == {} assert result[0] == {}
def test_check_preset_no_presets(self): def test_check_preset_no_presets(self):
@ -881,8 +876,7 @@ class TestNotification:
with patch("app.library.Notifications.NotificationEvents.is_valid", return_value=False): with patch("app.library.Notifications.NotificationEvents.is_valid", return_value=False):
result = notification.emit(event, None) result = notification.emit(event, None)
# Should return None and not submit to background worker assert result is None, "Should return None and not submit to background worker"
assert result is None
mock_worker_instance.submit.assert_not_called() mock_worker_instance.submit.assert_not_called()
def test_emit_valid_event(self): def test_emit_valid_event(self):
@ -908,8 +902,7 @@ class TestNotification:
with patch("app.library.Notifications.NotificationEvents.is_valid", return_value=True): with patch("app.library.Notifications.NotificationEvents.is_valid", return_value=True):
result = notification.emit(event, None) result = notification.emit(event, None)
# Should return None but submit to background worker assert result is None, "Should return None but submit to background worker"
assert result is None
mock_worker_instance.submit.assert_called_once_with(notification.send, event) mock_worker_instance.submit.assert_called_once_with(notification.send, event)
@pytest.mark.asyncio @pytest.mark.asyncio

View file

@ -23,8 +23,7 @@ class TestPreset:
assert preset.cookies == "" assert preset.cookies == ""
assert preset.cli == "" assert preset.cli == ""
assert preset.default is False assert preset.default is False
# ID should be auto-generated UUID assert len(preset.id) == 36, "ID should be auto-generated UUID" # UUID4 string length
assert len(preset.id) == 36 # UUID4 string length
assert "-" in preset.id assert "-" in preset.id
def test_preset_creation_with_all_fields(self): def test_preset_creation_with_all_fields(self):
@ -228,8 +227,7 @@ class TestPresets:
assert len(presets._items) == 1 assert len(presets._items) == 1
loaded_preset = presets._items[0] loaded_preset = presets._items[0]
assert loaded_preset.name == "old_preset" assert loaded_preset.name == "old_preset"
# Should have migrated format to cli assert "best[height<=720]" in loaded_preset.cli, "Should have migrated format to cli"
assert "best[height<=720]" in loaded_preset.cli
assert "--format" in loaded_preset.cli assert "--format" in loaded_preset.cli
# Should have generated ID # Should have generated ID
assert loaded_preset.id is not None assert loaded_preset.id is not None
@ -322,8 +320,7 @@ class TestPresets:
all_presets = presets.get_all() all_presets = presets.get_all()
# Should include both default and custom presets assert len(all_presets) > 1, "Should include both default and custom presets"
assert len(all_presets) > 1
assert any(p.name == "custom" for p in all_presets) assert any(p.name == "custom" for p in all_presets)
assert any(p.default is True 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, # Note: The actual chmod might fail in test environment,
# but we're testing that the code attempts it # 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): def test_default_presets_validation(self):
"""Test that all default presets are valid.""" """Test that all default presets are valid."""
@ -519,8 +516,7 @@ class TestPresets:
with patch.object(presets, "get", return_value=None): with patch.object(presets, "get", return_value=None):
# The actual config instance stored in presets should be checked # The actual config instance stored in presets should be checked
presets.attach(mock_app) presets.attach(mock_app)
# Should have reset default_preset to "default" assert presets._config.default_preset == "default", "Should have reset default_preset to 'default'"
assert presets._config.default_preset == "default"
@pytest.mark.asyncio @pytest.mark.asyncio
@patch("app.library.Presets.Config") @patch("app.library.Presets.Config")
@ -567,8 +563,7 @@ class TestPresets:
presets = Presets(file=presets_file, config=mock_config_instance) presets = Presets(file=presets_file, config=mock_config_instance)
presets.load() presets.load()
# Should only load the valid preset assert len(presets._items) == 1, "Should only load the valid preset"
assert len(presets._items) == 1
assert presets._items[0].name == "valid_preset", f"Expected 'valid_preset', got '{presets._items}'" assert presets._items[0].name == "valid_preset", f"Expected 'valid_preset', got '{presets._items}'"
# Should have logged an error for the invalid preset # Should have logged an error for the invalid preset
@ -658,8 +653,7 @@ class TestPresets:
all_presets = presets.get_all() all_presets = presets.get_all()
non_default = [p for p in all_presets if not p.default] non_default = [p for p in all_presets if not p.default]
# Should be sorted by priority descending assert non_default[0].name == "high", "Should be sorted by priority descending"
assert non_default[0].name == "high"
assert non_default[0].priority == 10 assert non_default[0].priority == 10
assert non_default[1].name == "medium" assert non_default[1].name == "medium"
assert non_default[1].priority == 5 assert non_default[1].priority == 5

View file

@ -130,8 +130,7 @@ class TestScheduler:
assert sched.has("job1") is True assert sched.has("job1") is True
new_job = sched.get("job1") new_job = sched.get("job1")
assert new_job is not old assert new_job is not old
# Old job should have been stopped via remove() assert old.stopped is True, "Old job should have been stopped via remove()"
assert old.stopped is True
@patch("app.library.Scheduler.Cron", new=DummyCron) @patch("app.library.Scheduler.Cron", new=DummyCron)
def test_remove_single_job_success(self) -> None: def test_remove_single_job_success(self) -> None:
@ -154,8 +153,7 @@ class TestScheduler:
result = sched.remove("jobB") result = sched.remove("jobB")
assert result is False assert result is False
# Job should remain since stop failed assert sched.has("jobB") is True, "Job should remain since stop failed"
assert sched.has("jobB") is True
@patch("app.library.Scheduler.Cron", new=DummyCron) @patch("app.library.Scheduler.Cron", new=DummyCron)
def test_remove_list_of_jobs(self) -> None: def test_remove_list_of_jobs(self) -> None:
@ -193,8 +191,9 @@ class TestScheduler:
sched = Scheduler() sched = Scheduler()
sched.attach(app) 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 # Patch add to verify it is called from event handler
add_spy = MagicMock(wraps=sched.add) add_spy = MagicMock(wraps=sched.add)

View file

@ -13,12 +13,13 @@ class TestMsToTimestamp:
assert ms_to_timestamp(9) == "0:00:00.00" assert ms_to_timestamp(9) == "0:00:00.00"
assert ms_to_timestamp(10) == "0:00:00.01" assert ms_to_timestamp(10) == "0:00:00.01"
assert ms_to_timestamp(12345) == "0:00:12.34" 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", "1 hour, 2 minutes, 3 seconds"
assert ms_to_timestamp(3600000 + 120000 + 3000) == "1:02:03.00" 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) "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(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 @pytest.mark.asyncio
@ -77,8 +78,7 @@ async def test_make_single_event_returns_vtt(tmp_path: Path) -> None:
sub = Subtitle() sub = Subtitle()
out = await sub.make(srt) out = await sub.make(srt)
assert out == "OUT" assert out == "OUT"
# Snapshot should contain the single event assert d.snapshot == [1000], "Snapshot should contain the single event"
assert d.snapshot == [1000]
@pytest.mark.asyncio @pytest.mark.asyncio
@ -94,8 +94,7 @@ async def test_make_two_events_pop_first_when_ends_equal(tmp_path: Path) -> None
sub = Subtitle() sub = Subtitle()
out = await sub.make(srt) out = await sub.make(srt)
assert out == "OUT" assert out == "OUT"
# Since ends are equal, first should be popped => only last remains assert d.snapshot == [5000], "Since ends are equal, first should be popped => only last remains"
assert d.snapshot == [5000]
@pytest.mark.asyncio @pytest.mark.asyncio
@ -111,5 +110,4 @@ async def test_make_two_events_no_pop_when_different(tmp_path: Path) -> None:
sub = Subtitle() sub = Subtitle()
out = await sub.make(srt) out = await sub.make(srt)
assert out == "OUT" assert out == "OUT"
# Both remain since ends differ assert d.snapshot == [5000, 6000], "Both remain since ends differ"
assert d.snapshot == [5000, 6000]

View file

@ -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"

View file

@ -321,8 +321,7 @@ class TestTasks:
tasks_file = Path(temp_dir) / "tasks.json" tasks_file = Path(temp_dir) / "tasks.json"
tasks = Tasks(file=tasks_file, config=mock_config_instance) tasks = Tasks(file=tasks_file, config=mock_config_instance)
# Check initialization assert tasks._debug is True, "Check initialization"
assert tasks._debug is True
assert tasks._default_preset == "test_preset" assert tasks._default_preset == "test_preset"
assert tasks._file == tasks_file assert tasks._file == tasks_file
assert tasks._tasks == [] assert tasks._tasks == []
@ -505,8 +504,7 @@ class TestTasks:
assert result == tasks assert result == tasks
assert tasks_file.exists() assert tasks_file.exists()
# Verify file content was written assert tasks_file.stat().st_size > 0, "Verify file content was written"
assert tasks_file.stat().st_size > 0
@patch("app.library.Tasks.Config") @patch("app.library.Tasks.Config")
@patch("app.library.Tasks.EventBus") @patch("app.library.Tasks.EventBus")
@ -788,8 +786,7 @@ class TestTasks:
} }
) )
# Verify events were emitted assert mock_eventbus_instance.emit.call_count == 2, "Verify events were emitted"
assert mock_eventbus_instance.emit.call_count == 2
@pytest.mark.asyncio @pytest.mark.asyncio
@patch("app.library.Tasks.Config") @patch("app.library.Tasks.Config")
@ -963,8 +960,7 @@ class TestHandleTaskInspect:
url="https://example.com/feed", preset="default", handler_name=None, static_only=True url="https://example.com/feed", preset="default", handler_name=None, static_only=True
) )
# Verify result structure assert hasattr(result, "items"), "Verify result structure"
assert hasattr(result, "items")
assert hasattr(result, "metadata") assert hasattr(result, "metadata")
assert result.items == [] assert result.items == []
assert result.metadata["matched"] is True assert result.metadata["matched"] is True
@ -1018,8 +1014,7 @@ class TestHandleTaskInspect:
# Call inspect with static_only=False (default) # Call inspect with static_only=False (default)
result = await handler.inspect(url="https://example.com/feed", preset="default", handler_name=None) result = await handler.inspect(url="https://example.com/feed", preset="default", handler_name=None)
# Verify result structure includes extracted items assert hasattr(result, "items"), "Verify result structure includes extracted items"
assert hasattr(result, "items")
assert hasattr(result, "metadata") assert hasattr(result, "metadata")
assert len(result.items) > 0 assert len(result.items) > 0
assert result.metadata["matched"] is True assert result.metadata["matched"] is True

View file

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

View file

@ -34,7 +34,7 @@ from app.library.Utils import (
get_files, get_files,
get_mime_type, get_mime_type,
get_possible_images, get_possible_images,
get_static_ytdlp, get_ytdlp,
init_class, init_class,
is_private_address, is_private_address,
list_folders, list_folders,
@ -137,9 +137,8 @@ class TestTimedLruCache:
def test_function(x): def test_function(x):
return x * 2 return x * 2
# Test that methods exist assert hasattr(test_function, "cache_clear"), "Cached function should have cache_clear method"
assert hasattr(test_function, "cache_clear") assert hasattr(test_function, "cache_info"), "Cached function should have cache_info method"
assert hasattr(test_function, "cache_info")
# Call function to populate cache # Call function to populate cache
test_function(5) test_function(5)
@ -256,9 +255,8 @@ class TestAsyncTimedLruCache:
async def async_method_test(x): async def async_method_test(x):
return x + 1 return x + 1
# Test that cache methods exist assert hasattr(async_method_test, "cache_clear"), "Async cached function should have cache_clear method"
assert hasattr(async_method_test, "cache_clear") assert hasattr(async_method_test, "cache_info"), "Async cached function should have cache_info method"
assert hasattr(async_method_test, "cache_info")
# Test cache_info # Test cache_info
info = async_method_test.cache_info() info = async_method_test.cache_info()
@ -286,12 +284,11 @@ class TestAsyncTimedLruCache:
# Fill cache beyond max_size # Fill cache beyond max_size
result1 = await async_limited_func(1) result1 = await async_limited_func(1)
result2 = await async_limited_func(2) 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, "async_limited_func(1) should return 4"
assert result1 == 4 assert result2 == 8, "async_limited_func(2) should return 8"
assert result2 == 8 assert result3 == 12, "async_limited_func(3) should return 12 (should evict oldest entry)"
assert result3 == 12
# Check cache size is limited # Check cache size is limited
info = async_limited_func.cache_info() info = async_limited_func.cache_info()
@ -664,9 +661,8 @@ class TestMergeDict:
source = {"nested": {"a": 1}} source = {"nested": {"a": 1}}
destination = {"nested": {"b": 2}, "other": 3} destination = {"nested": {"b": 2}, "other": 3}
result = merge_dict(source, destination) result = merge_dict(source, destination)
# Should merge nested dictionaries assert "nested" in result, "Should merge nested dictionaries"
assert "nested" in result assert "other" in result, "Should preserve other keys"
assert "other" in result
def test_merge_dict_empty_source(self): def test_merge_dict_empty_source(self):
"""Test merging with empty source.""" """Test merging with empty source."""
@ -695,7 +691,7 @@ class TestMergeDict:
destination = {"existing": "data"} destination = {"existing": "data"}
result = merge_dict(source, destination) 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["safe"] == "value", "Safe values should be preserved"
assert result["existing"] == "data", "Existing data 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 # All dangerous attributes should be filtered out
dangerous_keys = ["__class__", "__dict__", "__globals__", "__builtins__"] dangerous_keys = ["__class__", "__dict__", "__globals__", "__builtins__"]
for key in dangerous_keys: 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", "Safe data should be preserved"
assert result["safe_key"] == "safe_value" assert result["existing"] == "data", "Existing data should be preserved"
assert result["existing"] == "data"
def test_merge_dict_nested_dunder_pollution(self): def test_merge_dict_nested_dunder_pollution(self):
"""Test that nested dangerous attributes are handled correctly.""" """Test that nested dangerous attributes are handled correctly."""
@ -754,11 +749,9 @@ class TestMergeDict:
destination = {"nested": {"existing_nested": "original"}} destination = {"nested": {"existing_nested": "original"}}
result = merge_dict(source, destination) result = merge_dict(source, destination)
# Nested dangerous attributes should be filtered out assert "__class__" not in result["nested"], "Nested dangerous attributes should be filtered out"
assert "__class__" not in result["nested"], "Nested __class__ should be filtered" assert result["nested"]["safe_nested"] == "value", "Safe nested data should be preserved"
# Safe nested data should be preserved assert result["nested"]["existing_nested"] == "original", "Existing nested data should be preserved"
assert result["nested"]["safe_nested"] == "value"
assert result["nested"]["existing_nested"] == "original"
def test_merge_dict_prototype_pollution_attempt(self): def test_merge_dict_prototype_pollution_attempt(self):
"""Test protection against prototype pollution attempts.""" """Test protection against prototype pollution attempts."""
@ -766,10 +759,10 @@ class TestMergeDict:
destination = {"existing": "value"} destination = {"existing": "value"}
result = merge_dict(source, destination) result = merge_dict(source, destination)
# These should be treated as regular keys (not filtered unless explicitly in the filter list) assert result["safe"] == "data", (
# The function filters Python-specific dangerous attributes, not JavaScript ones "Function filters Python-specific dangerous attributes, not JS ones like __proto__"
assert result["safe"] == "data" )
assert result["existing"] == "value" assert result["existing"] == "value", "Existing data should be preserved"
def test_merge_dict_special_method_pollution(self): def test_merge_dict_special_method_pollution(self):
"""Test with various Python special methods.""" """Test with various Python special methods."""
@ -784,10 +777,10 @@ class TestMergeDict:
destination = {"target": "data"} destination = {"target": "data"}
result = merge_dict(source, destination) result = merge_dict(source, destination)
# These are not in the current filter list, so they should pass through assert result["safe"] == "value", (
# This test documents current behavior - may need updating if more filters are added "Safe data should be preserved (special methods not in filter list, documents current behavior)"
assert result["safe"] == "value" )
assert result["target"] == "data" assert result["target"] == "data", "Target data should be preserved"
def test_merge_dict_list_pollution_safe(self): def test_merge_dict_list_pollution_safe(self):
"""Test that list merging doesn't allow dangerous manipulation.""" """Test that list merging doesn't allow dangerous manipulation."""
@ -795,8 +788,9 @@ class TestMergeDict:
destination = {"items": ["old1", "old2"]} destination = {"items": ["old1", "old2"]}
result = merge_dict(source, destination) 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): def test_merge_dict_deep_nested_pollution(self):
"""Test with deeply nested dangerous attributes.""" """Test with deeply nested dangerous attributes."""
@ -812,13 +806,13 @@ class TestMergeDict:
destination = {"level1": {"level2": {"existing": "data"}}} destination = {"level1": {"level2": {"existing": "data"}}}
result = merge_dict(source, destination) result = merge_dict(source, destination)
# The function should now properly filter all dangerous keys recursively assert "__class__" not in result["level1"]["level2"], (
assert "__class__" not in result["level1"]["level2"], "Deep __class__ should be filtered" "Function should properly filter all dangerous keys recursively (deep __class__)"
assert "__globals__" not in result["level1"]["level2"]["level3"], "Very deep __globals__ should be filtered" )
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", "Safe nested data should be preserved"
assert result["level1"]["level2"]["safe_deep"] == "value" assert result["level1"]["level2"]["existing"] == "data", "Existing nested data should be preserved"
assert result["level1"]["level2"]["existing"] == "data"
def test_merge_dict_type_validation(self): def test_merge_dict_type_validation(self):
"""Test that non-dict parameters are properly rejected.""" """Test that non-dict parameters are properly rejected."""
@ -845,13 +839,13 @@ class TestMergeDict:
result = merge_dict(original_source, original_destination) result = merge_dict(original_source, original_destination)
# Original dictionaries should be unchanged assert original_source == source_copy, "Original source dictionary should be unchanged (immutability)"
assert original_source == source_copy, "Source should not be modified" assert original_destination == destination_copy, (
assert original_destination == destination_copy, "Destination should not be modified" "Original destination dictionary should be unchanged (immutability)"
)
# Result should be different from both originals assert result != original_source, "Result should be different from source original"
assert result != original_source assert result != original_destination, "Result should be different from destination original"
assert result != original_destination
def test_merge_dict_custom_max_depth(self): def test_merge_dict_custom_max_depth(self):
"""Test custom max_depth parameter.""" """Test custom max_depth parameter."""
@ -900,16 +894,14 @@ class TestMergeDict:
source = {"items": list(range(3000))} source = {"items": list(range(3000))}
destination = {"items": list(range(2000, 5000))} # 3000 items 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) 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)), "First 3000 items should be from destination"
assert result["items"][:3000] == list(range(2000, 5000)) assert result["items"][3000:] == list(range(1000)), "Next 1000 items should be from source (truncated)"
# Next 1000 should be from source (truncated)
assert result["items"][3000:] == list(range(1000))
def test_merge_dict_nested_with_limits(self): def test_merge_dict_nested_with_limits(self):
"""Test nested merging with both depth and size limits.""" """Test nested merging with both depth and size limits."""
@ -2018,8 +2010,7 @@ class TestDecryptDataCornerCases:
encrypted = encrypt_data("test data", correct_key) encrypted = encrypt_data("test data", correct_key)
result = decrypt_data(encrypted, wrong_key) result = decrypt_data(encrypted, wrong_key)
# Should return None on decryption failure assert result is None, "Should return None on decryption failure with wrong key"
assert result is None
def test_decrypt_data_truncated(self): def test_decrypt_data_truncated(self):
"""Test decrypting truncated encrypted data.""" """Test decrypting truncated encrypted data."""
@ -2175,11 +2166,10 @@ class TestRenameFile:
# Rename file # Rename file
new_path, sidecars = rename_file(test_file, "renamed_video.mp4") new_path, sidecars = rename_file(test_file, "renamed_video.mp4")
# Assertions assert new_path.exists(), "Renamed file should exist"
assert new_path.exists() assert "renamed_video.mp4" == new_path.name, "File should have new name"
assert "renamed_video.mp4" == new_path.name assert not test_file.exists(), "Original file should not exist"
assert not test_file.exists() assert 0 == len(sidecars), "Should have no sidecar files"
assert 0 == len(sidecars)
def test_rename_file_with_subtitle_sidecar(self, tmp_path: Path): def test_rename_file_with_subtitle_sidecar(self, tmp_path: Path):
"""Test renaming a file with subtitle sidecar.""" """Test renaming a file with subtitle sidecar."""
@ -2193,12 +2183,11 @@ class TestRenameFile:
# Rename file # Rename file
new_path, sidecars = rename_file(test_file, "renamed_video.mp4") new_path, sidecars = rename_file(test_file, "renamed_video.mp4")
# Assertions assert new_path.exists(), "Renamed file should exist"
assert new_path.exists() assert "renamed_video.mp4" == new_path.name, "File should have new name"
assert "renamed_video.mp4" == new_path.name assert not test_file.exists(), "Original file should not exist after rename"
assert not test_file.exists()
assert 1 == len(sidecars) assert 1 == len(sidecars), "Should have renamed 1 sidecar file"
old_sidecar, new_sidecar = sidecars[0] old_sidecar, new_sidecar = sidecars[0]
assert new_sidecar.exists() assert new_sidecar.exists()
assert "renamed_video.en.srt" == new_sidecar.name assert "renamed_video.en.srt" == new_sidecar.name
@ -2223,12 +2212,11 @@ class TestRenameFile:
# Rename file # Rename file
new_path, sidecars = rename_file(test_file, "renamed_video.mp4") new_path, sidecars = rename_file(test_file, "renamed_video.mp4")
# Assertions assert new_path.exists(), "Renamed file should exist"
assert new_path.exists() assert "renamed_video.mp4" == new_path.name, "File should have new name"
assert "renamed_video.mp4" == new_path.name assert not test_file.exists(), "Original file should not exist after rename"
assert not test_file.exists()
assert 3 == len(sidecars) assert 3 == len(sidecars), "Should have renamed 3 sidecar files"
# Check all sidecars were renamed # Check all sidecars were renamed
sidecar_names = {new_sidecar.name for old_sidecar, new_sidecar in sidecars} 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.fr.srt" in sidecar_names
assert "renamed_video.info.json" in sidecar_names assert "renamed_video.info.json" in sidecar_names
# Check old files don't exist assert not subtitle_en.exists(), "Old subtitle file should not exist after rename"
assert not subtitle_en.exists() assert not subtitle_fr.exists(), "Old subtitle file should not exist after rename"
assert not subtitle_fr.exists() assert not info_file.exists(), "Old info file should not exist after rename"
assert not info_file.exists()
def test_rename_file_destination_exists(self, tmp_path: Path): def test_rename_file_destination_exists(self, tmp_path: Path):
"""Test renaming a file when destination already exists.""" """Test renaming a file when destination already exists."""
@ -2254,9 +2241,8 @@ class TestRenameFile:
with pytest.raises(ValueError, match="already exists"): with pytest.raises(ValueError, match="already exists"):
rename_file(test_file, "renamed_video.mp4") rename_file(test_file, "renamed_video.mp4")
# Original files should still exist assert test_file.exists(), "Original file should still exist when rename fails"
assert test_file.exists() assert existing_file.exists(), "Existing file should still exist when rename fails"
assert existing_file.exists()
def test_rename_file_sidecar_destination_exists(self, tmp_path: Path): def test_rename_file_sidecar_destination_exists(self, tmp_path: Path):
"""Test renaming when sidecar destination already exists.""" """Test renaming when sidecar destination already exists."""
@ -2295,11 +2281,10 @@ class TestRenameFile:
# Rename file # Rename file
new_path, sidecars = rename_file(test_file, "renamed.mp4") new_path, sidecars = rename_file(test_file, "renamed.mp4")
# Assertions assert new_path.exists(), "Renamed file should exist"
assert new_path.exists() assert "renamed.mp4" == new_path.name, "File should have new name"
assert "renamed.mp4" == new_path.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} sidecar_names = {new_sidecar.name for old_sidecar, new_sidecar in sidecars}
assert "renamed.en-US.ass" in sidecar_names assert "renamed.en-US.ass" in sidecar_names
assert "renamed.thumb.jpg" in sidecar_names assert "renamed.thumb.jpg" in sidecar_names
@ -2322,12 +2307,11 @@ class TestMoveFile:
# Move file # Move file
new_path, sidecars = move_file(test_file, target_dir) new_path, sidecars = move_file(test_file, target_dir)
# Assertions assert new_path.exists(), "Moved file should exist at destination"
assert new_path.exists() assert "video.mp4" == new_path.name, "File should keep same name"
assert "video.mp4" == new_path.name assert new_path.parent == target_dir, "File should be in target directory"
assert new_path.parent == target_dir assert not test_file.exists(), "Original file should not exist after move"
assert not test_file.exists() assert 0 == len(sidecars), "Should have no sidecar files"
assert 0 == len(sidecars)
def test_move_file_with_subtitle_sidecar(self, tmp_path: Path): def test_move_file_with_subtitle_sidecar(self, tmp_path: Path):
"""Test moving a file with subtitle sidecar.""" """Test moving a file with subtitle sidecar."""
@ -2346,13 +2330,12 @@ class TestMoveFile:
# Move file # Move file
new_path, sidecars = move_file(test_file, target_dir) new_path, sidecars = move_file(test_file, target_dir)
# Assertions assert new_path.exists(), "Moved file should exist at destination"
assert new_path.exists() assert "video.mp4" == new_path.name, "File should keep same name"
assert "video.mp4" == new_path.name assert new_path.parent == target_dir, "File should be in target directory"
assert new_path.parent == target_dir assert not test_file.exists(), "Original file should not exist after move"
assert not test_file.exists()
assert 1 == len(sidecars) assert 1 == len(sidecars), "Should have moved 1 sidecar file"
old_sidecar, new_sidecar = sidecars[0] old_sidecar, new_sidecar = sidecars[0]
assert new_sidecar.exists() assert new_sidecar.exists()
assert "video.en.srt" == new_sidecar.name assert "video.en.srt" == new_sidecar.name
@ -2383,13 +2366,12 @@ class TestMoveFile:
# Move file # Move file
new_path, sidecars = move_file(test_file, target_dir) new_path, sidecars = move_file(test_file, target_dir)
# Assertions assert new_path.exists(), "Moved file should exist at destination"
assert new_path.exists() assert "video.mp4" == new_path.name, "File should keep same name"
assert "video.mp4" == new_path.name assert new_path.parent == target_dir, "File should be in target directory"
assert new_path.parent == target_dir assert not test_file.exists(), "Original file should not exist after move"
assert not test_file.exists()
assert 3 == len(sidecars) assert 3 == len(sidecars), "Should have moved 3 sidecar files"
# Check all sidecars were moved # Check all sidecars were moved
sidecar_names = {new_sidecar.name for old_sidecar, new_sidecar in sidecars} sidecar_names = {new_sidecar.name for old_sidecar, new_sidecar in sidecars}
@ -2399,12 +2381,11 @@ class TestMoveFile:
# Check all are in target directory # Check all are in target directory
for _old_sidecar, new_sidecar in sidecars: 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(), "Old subtitle file should not exist after move"
assert not subtitle_en.exists() assert not subtitle_fr.exists(), "Old subtitle file should not exist after move"
assert not subtitle_fr.exists() assert not info_file.exists(), "Old info file should not exist after move"
assert not info_file.exists()
def test_move_file_destination_exists(self, tmp_path: Path): def test_move_file_destination_exists(self, tmp_path: Path):
"""Test moving a file when destination already exists.""" """Test moving a file when destination already exists."""
@ -2423,9 +2404,8 @@ class TestMoveFile:
with pytest.raises(ValueError, match="already exists"): with pytest.raises(ValueError, match="already exists"):
move_file(test_file, target_dir) move_file(test_file, target_dir)
# Original files should still exist assert test_file.exists(), "Original file should still exist when move fails"
assert test_file.exists() assert existing_file.exists(), "Existing file should still exist when move fails"
assert existing_file.exists()
def test_move_file_sidecar_destination_exists(self, tmp_path: Path): def test_move_file_sidecar_destination_exists(self, tmp_path: Path):
"""Test moving when sidecar destination already exists.""" """Test moving when sidecar destination already exists."""
@ -2682,41 +2662,43 @@ class TestGetExtras:
class TestGetStaticYtdlp: class TestGetStaticYtdlp:
"""Test the get_static_ytdlp function.""" """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): def test_get_static_ytdlp_returns_instance(self):
"""Test that get_static_ytdlp returns a YTDLP instance.""" """Test that get_static_ytdlp returns a YTDLP instance."""
from app.library.Utils import get_static_ytdlp
from app.library.ytdlp import YTDLP from app.library.ytdlp import YTDLP
# Force reload to ensure we get a real instance, not a mock # Get the cached instance
instance = get_static_ytdlp(reload=True) instance = get_ytdlp()
assert instance is not None assert instance is not None
assert isinstance(instance, YTDLP) assert isinstance(instance, YTDLP)
def test_get_static_ytdlp_returns_same_instance(self): def test_get_static_ytdlp_returns_same_instance(self):
"""Test that get_static_ytdlp returns the same cached instance.""" """Test that get_static_ytdlp returns the same cached instance."""
from app.library.Utils import get_static_ytdlp
instance1 = get_static_ytdlp() instance1 = get_ytdlp()
instance2 = get_static_ytdlp() instance2 = get_ytdlp()
assert instance1 is instance2 assert instance1 is instance2
def test_get_static_ytdlp_reload(self): def test_get_static_ytdlp_with_params(self):
"""Test that get_static_ytdlp can reload and return a new instance.""" """Test that get_static_ytdlp returns a new instance when params are provided."""
from app.library.Utils import get_static_ytdlp
instance1 = get_static_ytdlp() instance1 = get_ytdlp()
instance2 = get_static_ytdlp(reload=True) instance2 = get_ytdlp(params={"quiet": False})
assert instance1 is not instance2 assert instance1 is not instance2
assert instance2 is not None assert instance2 is not None
def test_get_static_ytdlp_has_correct_params(self): def test_get_static_ytdlp_has_correct_params(self):
"""Test that get_static_ytdlp initializes with correct parameters.""" """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 # Access the internal params
params = instance.params params = instance.params
@ -2732,9 +2714,14 @@ class TestGetStaticYtdlp:
class TestParseOuttmpl: class TestParseOuttmpl:
"""Test the parse_outtmpl function.""" """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): def test_parse_outtmpl_basic(self):
"""Test basic template parsing with simple placeholders.""" """Test basic template parsing with simple placeholders."""
from app.library.Utils import parse_outtmpl
template = "%(title)s.%(ext)s" template = "%(title)s.%(ext)s"
info_dict = { info_dict = {
@ -2748,7 +2735,6 @@ class TestParseOuttmpl:
def test_parse_outtmpl_with_id(self): def test_parse_outtmpl_with_id(self):
"""Test template parsing with video ID.""" """Test template parsing with video ID."""
from app.library.Utils import parse_outtmpl
template = "[%(id)s] %(title)s.%(ext)s" template = "[%(id)s] %(title)s.%(ext)s"
info_dict = { info_dict = {
@ -2763,7 +2749,6 @@ class TestParseOuttmpl:
def test_parse_outtmpl_with_uploader(self): def test_parse_outtmpl_with_uploader(self):
"""Test template parsing with uploader information.""" """Test template parsing with uploader information."""
from app.library.Utils import parse_outtmpl
template = "%(uploader)s - %(title)s.%(ext)s" template = "%(uploader)s - %(title)s.%(ext)s"
info_dict = { info_dict = {
@ -2778,7 +2763,6 @@ class TestParseOuttmpl:
def test_parse_outtmpl_with_nested_path(self): def test_parse_outtmpl_with_nested_path(self):
"""Test template parsing with nested directory structure.""" """Test template parsing with nested directory structure."""
from app.library.Utils import parse_outtmpl
template = "%(uploader)s/%(title)s.%(ext)s" template = "%(uploader)s/%(title)s.%(ext)s"
info_dict = { info_dict = {
@ -2793,22 +2777,19 @@ class TestParseOuttmpl:
def test_parse_outtmpl_with_missing_field(self): def test_parse_outtmpl_with_missing_field(self):
"""Test template parsing with missing field defaults to NA.""" """Test template parsing with missing field defaults to NA."""
from app.library.Utils import parse_outtmpl
template = "%(title)s - %(upload_date)s.%(ext)s" template = "%(title)s - %(upload_date)s.%(ext)s"
info_dict = { info_dict = {
"title": "Test Video", "title": "Test Video",
"ext": "mp4", "ext": "mp4",
# upload_date is missing
} }
result = parse_outtmpl(template, info_dict) 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): def test_parse_outtmpl_complex(self):
"""Test complex template with multiple fields.""" """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" template = "%(uploader)s/%(playlist_title)s/%(playlist_index)03d - %(title)s [%(id)s].%(ext)s"
info_dict = { info_dict = {
@ -2826,7 +2807,6 @@ class TestParseOuttmpl:
def test_parse_outtmpl_with_special_characters(self): def test_parse_outtmpl_with_special_characters(self):
"""Test template parsing handles special characters in values.""" """Test template parsing handles special characters in values."""
from app.library.Utils import parse_outtmpl
template = "%(title)s.%(ext)s" template = "%(title)s.%(ext)s"
info_dict = { info_dict = {
@ -2836,13 +2816,11 @@ class TestParseOuttmpl:
result = parse_outtmpl(template, info_dict) result = parse_outtmpl(template, info_dict)
# yt-dlp sanitizes special characters in filenames assert ".mp4" in result, "yt-dlp should sanitize special characters but preserve extension"
assert ".mp4" in result assert "Test" in result, "yt-dlp should preserve safe parts of title"
assert "Test" in result
def test_parse_outtmpl_with_playlist_info(self): def test_parse_outtmpl_with_playlist_info(self):
"""Test template parsing with playlist information.""" """Test template parsing with playlist information."""
from app.library.Utils import parse_outtmpl
template = "%(playlist)s/%(title)s.%(ext)s" template = "%(playlist)s/%(title)s.%(ext)s"
info_dict = { info_dict = {
@ -2854,3 +2832,19 @@ class TestParseOuttmpl:
result = parse_outtmpl(template, info_dict) result = parse_outtmpl(template, info_dict)
assert result == "My Playlist/Video Title.webm" 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"

View file

@ -75,8 +75,7 @@ class TestYtDlpOptions:
# For any ignored flag that actually exists in yt-dlp parser, ensure it is marked ignored # 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] 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, "We expect at least one to be present (e.g., -P / --paths, etc.)"
assert len(present_ignored_flags) > 0
assert all(flag_to_ignored[f] is True for f in present_ignored_flags) 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() # Our __init__ code manually sets these after super()
ytdlp.params["download_archive"] = "/tmp/archive.txt" ytdlp.params["download_archive"] = "/tmp/archive.txt"
# Verify download_archive was restored to params assert ytdlp.params["download_archive"] == "/tmp/archive.txt", "Verify download_archive was restored to params"
assert ytdlp.params["download_archive"] == "/tmp/archive.txt"
# Verify archive proxy was set up # Verify archive proxy was set up
assert isinstance(ytdlp.archive, _ArchiveProxy) assert isinstance(ytdlp.archive, _ArchiveProxy)
@ -133,8 +131,7 @@ class TestYTDLP:
assert call_kwargs["params"]["quiet"] is True assert call_kwargs["params"]["quiet"] is True
assert call_kwargs["auto_init"] is False assert call_kwargs["auto_init"] is False
# Verify archive proxy is falsey assert isinstance(ytdlp.archive, _ArchiveProxy), "Verify archive proxy is falsey"
assert isinstance(ytdlp.archive, _ArchiveProxy)
assert not ytdlp.archive assert not ytdlp.archive
@patch("app.library.ytdlp.yt_dlp.YoutubeDL.__init__") @patch("app.library.ytdlp.yt_dlp.YoutubeDL.__init__")
@ -230,8 +227,7 @@ class TestYTDLP:
assert ytdlp.archive.add.call_count == 3 assert ytdlp.archive.add.call_count == 3
calls = [call[0][0] for call in ytdlp.archive.add.call_args_list] calls = [call[0][0] for call in ytdlp.archive.add.call_args_list]
# First call is main archive_id assert calls[0] == "youtube new123", "First call is main archive_id"
assert calls[0] == "youtube new123"
# Should add old IDs except the duplicate # Should add old IDs except the duplicate
assert "youtube old123" in calls assert "youtube old123" in calls

View file

@ -474,8 +474,7 @@ class TestYTDLPOpts:
assert "Failed to load" in error_args assert "Failed to load" in error_args
assert "Cookie Preset" in error_args assert "Cookie Preset" in error_args
# cookiefile should not be set assert "cookiefile" not in opts._preset_opts, "cookiefile should not be set"
assert "cookiefile" not in opts._preset_opts
def test_replacer_substitution_in_cli(self): def test_replacer_substitution_in_cli(self):
"""Test that CLI arguments get replacer substitution.""" """Test that CLI arguments get replacer substitution."""
@ -540,13 +539,11 @@ class TestARGSMerger:
merger.add(cli_with_comments) merger.add(cli_with_comments)
# Comments should be filtered out assert "#" not in merger.as_string(), "Comments should be filtered out"
assert "#" not in merger.as_string()
assert "This is a comment" not in merger.as_string() assert "This is a comment" not in merger.as_string()
assert "Another comment" not in merger.as_string() assert "Another comment" not in merger.as_string()
# Valid options should remain assert "--format" in merger.args, "Valid options should remain"
assert "--format" in merger.args
assert "best" in merger.args assert "best" in merger.args
assert "--output" in merger.args assert "--output" in merger.args
assert "test.mp4" in merger.args assert "test.mp4" in merger.args
@ -566,14 +563,12 @@ class TestARGSMerger:
merger.add(cli_with_indented_comments) merger.add(cli_with_indented_comments)
# Comments should be filtered out
result = merger.as_string() 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 "# Indented comment with tabs" not in result
assert "# Another indented comment" not in result assert "# Another indented comment" not in result
# Valid options should remain assert "--format" in merger.args, "Valid options should remain"
assert "--format" in merger.args
assert "--output" in merger.args assert "--output" in merger.args
assert "--socket-timeout" in merger.args assert "--socket-timeout" in merger.args
@ -591,13 +586,14 @@ class TestARGSMerger:
result = merger.as_string() result = merger.as_string()
# Commented lines should be filtered out completely assert "player_js_version=actual" not in result, "Commented lines should be filtered out completely"
assert "player_js_version=actual" not in result assert "mweb,web_safari;formats=incomplete" not in result, (
# Check the specific commented variant (with comma before web_safari, not dash) "Check the specific commented variant (with comma before web_safari, not dash)"
assert "mweb,web_safari;formats=incomplete" not in result )
# 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 "--socket-timeout" in merger.args
assert "60" in merger.args assert "60" in merger.args
@ -839,8 +835,7 @@ class TestYTDLPCli:
cli = YTDLPCli(item=item) cli = YTDLPCli(item=item)
command, info = cli.build() command, info = cli.build()
# Should use preset values assert info["merged"]["template"] == "%(channel)s/%(title)s.%(ext)s", "Should use preset values"
assert info["merged"]["template"] == "%(channel)s/%(title)s.%(ext)s"
assert info["merged"]["save_path"] == "/downloads/preset_folder" assert info["merged"]["save_path"] == "/downloads/preset_folder"
assert "--format 720p" in command assert "--format 720p" in command
@ -879,11 +874,9 @@ class TestYTDLPCli:
cli = YTDLPCli(item=item) cli = YTDLPCli(item=item)
command, info = cli.build() command, info = cli.build()
# Should use user values, not preset assert info["merged"]["template"] == "%(id)s.%(ext)s", "Should use user values, not preset"
assert info["merged"]["template"] == "%(id)s.%(ext)s"
assert info["merged"]["save_path"] == "/downloads/user_folder" assert info["merged"]["save_path"] == "/downloads/user_folder"
# User CLI should appear after preset CLI in command assert "--format best" in command, "User CLI should appear after preset CLI in command"
assert "--format best" in command
@patch("app.library.Presets.Presets") @patch("app.library.Presets.Presets")
@patch("app.library.YTDLPOpts.Config") @patch("app.library.YTDLPOpts.Config")
@ -998,8 +991,9 @@ class TestYTDLPCli:
cli = YTDLPCli(item=item) cli = YTDLPCli(item=item)
command, info = cli.build() 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 assert "--paths" in command
@patch("app.library.Presets.Presets") @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 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_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, "User's 'best' should appear after preset's '720p'"
assert user_format_idx > preset_format_idx

View file

@ -157,34 +157,113 @@
</ClientOnly> </ClientOnly>
</div> </div>
<div class="columns mt-3 is-mobile"> <footer class="footer py-5 mt-6 is-unselectable" v-if="socket.isConnected">
<div class="column"> <div class="columns is-multiline is-variable is-8">
<div class="has-text-left" v-if="config.app?.app_version"> <div class="column is-12-mobile is-6-tablet">
© {{ Year }} - <NuxtLink href="https://github.com/ArabCoders/ytptube" target="_blank">YTPTube</NuxtLink> <div class="mb-3">
(<span class="has-tooltip" <NuxtLink href="https://github.com/ArabCoders/ytptube" target="_blank"
v-tooltip="`Build Date: ${config.app?.app_build_date}, Branch: ${config.app?.app_branch}, commit: ${config.app?.app_commit_sha}`"> class="has-text-weight-semibold is-size-6">
{{ config?.app?.app_version || 'unknown' }}</span>) <span class="icon-text">
- <NuxtLink target="_blank" href="https://github.com/yt-dlp/yt-dlp">yt-dlp</NuxtLink> <span class="icon"><i class="fab fa-github" /></span>
<span>&nbsp;({{ config?.app?.ytdlp_version || 'unknown' }})</span> <span>YTPTube</span>
- <NuxtLink to="/changelog">CHANGELOG</NuxtLink> </span>
- <NuxtLink @click="doc.file = '/api/docs/FAQ.md'">FAQ</NuxtLink> </NuxtLink>
- <NuxtLink @click="doc.file = '/api/docs/README.md'">README</NuxtLink> <span class="is-size-7 ml-2 has-tooltip" style="opacity: 0.7"
- <NuxtLink @click="doc.file = '/api/docs/API.md'">API</NuxtLink> v-tooltip="`Build: ${config.app?.app_build_date}, Branch: ${config.app?.app_branch}, SHA: ${config.app?.app_commit_sha}`">
- <NuxtLink @click="scrollToTop"> {{ config?.app?.app_version || 'unknown' }}
<span class="icon"><i class="fas fa-arrow-up" /></span> </span>
<span>Top</span> </div>
</NuxtLink>
<template v-if="config.app?.check_for_updates">
<p v-if="config.app?.new_version" class="is-size-7 mb-2" style="opacity: 0.8">
<span class="icon-text">
<span class="icon has-text-warning"><i class="fas fa-circle-info" /></span>
<span>Update available:</span>
<NuxtLink :href="`https://github.com/ArabCoders/ytptube/releases/tag/${config.app.new_version}`"
target="_blank" class="has-text-weight-semibold ml-1">
{{ config.app.new_version }}
</NuxtLink>
</span>
</p>
<p v-else class="is-size-7 mb-2" style="opacity: 0.6">
<button @click="checkForUpdates" :disabled="checkingUpdates" class="is-text is-small p-0 is-size-7"
:class="{ 'is-loading': checkingUpdates }"
style="opacity: 0.8; height: auto; vertical-align: baseline; border: none; text-decoration: none; background: none;">
<span class="icon-text">
<span class="icon">
<i class="fas" :class="checkingUpdates ? 'fa-spinner fa-spin' : 'fa-circle-check'" />
</span>
<span>{{ updateCheckMessage }}</span>
</span>
</button>
</p>
</template>
<p v-if="config.app?.started" class="is-size-7 mb-0" style="opacity: 0.6">
<span class="icon-text">
<span class="icon"><i class="fas fa-clock" /></span>
<span class="has-tooltip"
v-tooltip="'Started: ' + moment.unix(config.app?.started).format('YYYY-MM-DD HH:mm Z')">
{{ moment.unix(config.app?.started).fromNow() }}
</span>
</span>
</p>
</div>
<div class="column is-12-mobile is-3-tablet">
<div class="footer-divider"
style="border-left: 1px solid rgba(128,128,128,0.2); border-right: 1px solid rgba(128,128,128,0.2); padding: 0 2rem;">
<p class="is-size-7 mb-2" style="opacity: 0.7">Powered by</p>
<NuxtLink href="https://github.com/yt-dlp/yt-dlp" target="_blank" class="has-text-weight-semibold">
<span class="icon-text">
<span class="icon"><i class="fab fa-github" /></span>
<span>yt-dlp</span>
</span>
</NuxtLink>
<span class="is-size-7 ml-1" style="opacity: 0.6">{{ config?.app?.ytdlp_version || 'unknown' }}</span>
</div>
</div>
<div class="column is-12-mobile is-3-tablet">
<p class="is-size-7 mb-2" style="opacity: 0.7">Quick Links</p>
<div
class="is-flex is-flex-direction-row is-flex-wrap-wrap is-justify-content-flex-start is-justify-content-flex-end-tablet"
style="gap: 0.75rem;">
<NuxtLink to="/changelog" class="is-size-7">
<span class="icon-text">
<span class="icon"><i class="fas fa-list" /></span>
<span>Changelog</span>
</span>
</NuxtLink>
<NuxtLink @click="doc.file = '/api/docs/FAQ.md'" class="is-size-7">
<span class="icon-text">
<span class="icon"><i class="fas fa-question-circle" /></span>
<span>FAQ</span>
</span>
</NuxtLink>
<NuxtLink @click="doc.file = '/api/docs/README.md'" class="is-size-7">
<span class="icon-text">
<span class="icon"><i class="fas fa-book" /></span>
<span>Docs</span>
</span>
</NuxtLink>
<NuxtLink @click="doc.file = '/api/docs/API.md'" class="is-size-7">
<span class="icon-text">
<span class="icon"><i class="fas fa-code" /></span>
<span>API</span>
</span>
</NuxtLink>
<button @click="scrollToTop" class="is-size-7 button is-text is-small p-0">
<span class="icon-text">
<span class="icon"><i class="fas fa-arrow-up" /></span>
<span>Top</span>
</span>
</button>
</div>
</div> </div>
</div> </div>
<div class="column is-narrow" v-if="config.app?.started"> </footer>
<div class="has-text-right">
<span class="has-tooltip"
v-tooltip="'App Started: ' + moment.unix(config.app?.started).format('YYYY-M-DD H:mm Z')">
{{ moment.unix(config.app?.started).fromNow() }}
</span>
</div>
</div>
</div>
</div> </div>
</template> </template>
</template> </template>
@ -204,7 +283,6 @@ import Shutdown from '~/components/shutdown.vue'
import Markdown from '~/components/Markdown.vue' import Markdown from '~/components/Markdown.vue'
import Connection from '~/components/Connection.vue' import Connection from '~/components/Connection.vue'
const Year = new Date().getFullYear()
const selectedTheme = useStorage('theme', 'auto') const selectedTheme = useStorage('theme', 'auto')
const socket = useSocketStore() const socket = useSocketStore()
const config = useConfigStore() const config = useConfigStore()
@ -218,6 +296,53 @@ const app_shutdown = ref<boolean>(false)
const simpleMode = useStorage<boolean>('simple_mode', config.app.simple_mode || false) const simpleMode = useStorage<boolean>('simple_mode', config.app.simple_mode || false)
const show_settings = ref(false) const show_settings = ref(false)
const doc = ref<{ file: string }>({ file: '' }) 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) => { const applyPreferredColorScheme = (scheme: string) => {
if (!scheme || scheme === 'auto') { if (!scheme || scheme === 'auto') {
@ -515,5 +640,14 @@ const scrollToTop = () => document.getElementById('top')?.scrollIntoView({ behav
.basic-wrapper.settings-open { .basic-wrapper.settings-open {
transform: translateX(0); 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;
}
} }
</style> </style>

View file

@ -408,25 +408,13 @@ const editItem = (item: Preset) => {
onMounted(async () => (socket.isConnected ? await reloadContent(true) : '')) onMounted(async () => (socket.isConnected ? await reloadContent(true) : ''))
const exportItem = (item: Preset) => { const exportItem = (item: Preset) => {
const keys = ['id', 'default', 'raw', 'cookies', 'toggle_description'] const excludedKeys = ['id', 'default', 'raw', 'cookies', 'toggle_description']
const data = JSON.parse(JSON.stringify(item)) const userData = Object.fromEntries(
Object.entries(JSON.parse(JSON.stringify(item))).filter(([key, value]) => !excludedKeys.includes(key) && value)
for (const key of keys) { )
if (key in data) {
const { [key]: _, ...rest } = data
Object.assign(data, rest)
}
}
const userData: Record<string, any> = {}
for (const key of Object.keys(data)) {
if (data[key]) {
userData[key] = data[key]
}
}
userData['_type'] = 'preset' userData['_type'] = 'preset'
userData['_version'] = '2.5' userData['_version'] = '2.6'
copyText(encode(userData)) copyText(encode(userData))
} }

View file

@ -26,6 +26,8 @@ export const useConfigStore = defineStore('config', () => {
app_env: 'production', app_env: 'production',
simple_mode: false, simple_mode: false,
default_pagination: 50, default_pagination: 50,
check_for_updates: true,
new_version: '',
}, },
presets: [ presets: [
{ {

View file

@ -45,6 +45,10 @@ type AppConfig = {
app_env: "production" | "development" app_env: "production" | "development"
/** Default number of items per page for pagination */ /** Default number of items per page for pagination */
default_pagination: number 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 = { type ConfigState = {