danvics/ytptube
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

fix linter issues

Browse source
This commit is contained in:
arabcoders 2025-09-11 16:34:11 +03:00
parent 669256fa69
commit 7a50cb34d8
40 changed files with 2816 additions and 142 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

59
.github/workflows/build-pr.yml vendored
View file

@ -1,4 +1,4 @@
name: Test Build PR
name: Test PR
permissions:
contents: read
@ -14,14 +14,37 @@ on:
env:
PNPM_VERSION: 10
NODE_VERSION: 20
PYTHON_VERSION: "3.11"
jobs:
build-pr:
test:
name: Run Tests & Build Validation
runs-on: ubuntu-latest
steps:
- name: Checkout
uses: actions/checkout@v4
# Python setup and testing
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: ${{ env.PYTHON_VERSION }}
- name: Install uv
uses: astral-sh/setup-uv@v4
with:
version: "latest"
- name: Install Python dependencies
run: uv sync
- name: Run Python linting
run: uv run ruff check app/
- name: Run Python tests
run: uv run pytest app/tests/ -v --tb=short
# Frontend setup and testing
- name: Install pnpm
uses: pnpm/action-setup@v4
with:
@ -34,16 +57,37 @@ jobs:
cache: pnpm
cache-dependency-path: "ui/pnpm-lock.yaml"
- name: Install frontend dependencies & Build
- name: Install frontend dependencies
working-directory: ui
run: |
pnpm install --production --prefer-offline --frozen-lockfile
pnpm run generate
run: pnpm install --frozen-lockfile
- name: Prepare frontend (nuxt prepare)
working-directory: ui
env:
NODE_ENV: development
run: pnpm nuxt prepare
- name: Run frontend linting
working-directory: ui
run: pnpm run lint
- name: Run frontend type checking
working-directory: ui
run: pnpm run typecheck
- name: Run frontend tests
working-directory: ui
run: pnpm run test:ci
- name: Build frontend
working-directory: ui
run: pnpm run generate
# Quick Docker build validation (no push, cache only)
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v3
- name: Build commit to check (amd64 only)
- name: Validate Docker build
uses: docker/build-push-action@v5
with:
context: .
@ -52,4 +96,3 @@ jobs:
cache-from: type=gha,scope=${{ github.workflow }}
cache-to: type=gha,mode=max,scope=${{ github.workflow }}
provenance: false

105
.github/workflows/main.yml vendored
View file

@ -26,29 +26,39 @@ env:
GHCR_SLUG: ghcr.io/arabcoders/ytptube
PNPM_VERSION: 10
NODE_VERSION: 20
PYTHON_VERSION: "3.11"
jobs:
docker-build-arch:
name: Build Container (${{ matrix.arch }})
runs-on: ${{ matrix.os }}
strategy:
matrix:
include:
- os: ubuntu-latest
arch: amd64
- os: ubuntu-latest
arch: arm64
permissions:
packages: write
contents: write
env:
ARCH: ${{ matrix.arch }}
test:
name: Run Tests & Prepare Frontend
runs-on: ubuntu-latest
steps:
- name: Checkout
uses: actions/checkout@v4
with:
fetch-depth: 0
# Python setup and testing
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: ${{ env.PYTHON_VERSION }}
- name: Install uv
uses: astral-sh/setup-uv@v4
with:
version: "latest"
- name: Install Python dependencies
run: uv sync
- name: Run Python linting
run: uv run ruff check app/
- name: Run Python tests
run: uv run pytest app/tests/ -v --tb=short
# Frontend setup and testing
- name: Install pnpm
uses: pnpm/action-setup@v4
with:
@ -61,12 +71,33 @@ jobs:
cache: pnpm
cache-dependency-path: "ui/pnpm-lock.yaml"
- name: Install frontend dependencies & Build
- name: Install frontend dependencies
working-directory: ui
run: |
pnpm install --production --prefer-offline --frozen-lockfile
pnpm run generate
run: pnpm install --frozen-lockfile
- name: Prepare frontend (nuxt prepare)
working-directory: ui
env:
NODE_ENV: development
run: pnpm nuxt prepare
- name: Run frontend linting
working-directory: ui
run: pnpm run lint
- name: Run frontend type checking
working-directory: ui
run: pnpm run typecheck
- name: Run frontend tests
working-directory: ui
run: pnpm run test:ci
- name: Build frontend
working-directory: ui
run: pnpm run generate
# Generate changelog and version files
- name: Install GitPython
run: pip install gitpython
@ -93,6 +124,42 @@ jobs:
-e "s/^APP_BRANCH = \".*\"/APP_BRANCH = \"${BRANCH}\"/" \
app/library/version.py
# Upload built frontend as artifact for docker builds
- name: Upload frontend build
uses: actions/upload-artifact@v4
with:
name: frontend-build
path: ui/exported/
retention-days: 1
docker-build-arch:
name: Build Container (${{ matrix.arch }})
runs-on: ${{ matrix.os }}
needs: [test]
strategy:
matrix:
include:
- os: ubuntu-latest
arch: amd64
- os: ubuntu-latest
arch: arm64
permissions:
packages: write
contents: write
env:
ARCH: ${{ matrix.arch }}
steps:
- name: Checkout
uses: actions/checkout@v4
with:
fetch-depth: 0
- name: Download frontend build
uses: actions/download-artifact@v4
with:
name: frontend-build
path: ui/exported/
- name: Docker meta
id: meta
uses: docker/metadata-action@v5

2
.gitignore vendored
View file

@ -37,3 +37,5 @@ __pycache__
dist
build
version.txt
test_impl.py
./eslint.config.js

2
.vscode/launch.json vendored
View file

@ -87,6 +87,6 @@
}
],
"justMyCode": true
}
},
]
}

17
.vscode/settings.json vendored
View file

@ -19,6 +19,7 @@
"anyio",
"Archiver",
"arrowless",
"asyncio",
"attl",
"autonumber",
"bgutil",
@ -41,8 +42,10 @@
"daterange",
"defusedxml",
"dlfields",
"dotdot",
"dotenv",
"dpkg",
"dunder",
"Edes",
"edgechromium",
"EISGPM",
@ -52,11 +55,14 @@
"euuo",
"excepthook",
"faststart",
"Fetc",
"finaldir",
"flac",
"forcejson",
"forceprint",
"Fpasswd",
"fribidi",
"Funsafe",
"getpid",
"gibibytes",
"gitpython",
@ -95,6 +101,7 @@
"noprogress",
"onefile",
"pathex",
"pickleable",
"platformdirs",
"playlistend",
"playlistrandom",
@ -121,12 +128,14 @@
"socketio",
"softprops",
"sstr",
"startswith",
"SUPPRESSHELP",
"tebibytes",
"testpaths",
"tiktok",
"timespec",
"tmpfilename",
"toplevel",
"ungroup",
"unmark",
"unnegated",
@ -139,10 +148,13 @@
"ustr",
"vcodec",
"vconvert",
"Vitest",
"writedescription",
"xerror",
"youtu",
"youtubepot"
"youtubepot",
"русский",
"العربية"
],
"css.styleSheets": [
"ui/app/assets/css/*.css"
@ -156,5 +168,6 @@
"ui/.out": true,
"**/.venv": true,
},
"eslint.format.enable": true
"eslint.format.enable": true,
"eslint.ignoreUntitled": true
}

96
app/library/Utils.py
View file

@ -226,38 +226,116 @@ def extract_info(
return YTDLP.sanitize_info(data) if sanitize_info else data
def merge_dict(source: dict, destination: dict) -> dict:
def _is_safe_key(key: any) -> bool:
"""
Merge data from source into destination safely.
Check if a dictionary key is safe for merging.
Blocks:
- All dunder attributes (__*__)
- Non-string keys (for consistency)
- Keys that could be dangerous variations
"""
# Only allow string keys
if not isinstance(key, str):
return False
# Block empty keys and whitespace-only keys
if not key or not key.strip():
return False
# Block only truly dangerous dunder patterns
key_stripped = key.strip()
# Block dunder attributes (starts AND ends with __)
return not (key_stripped.startswith("__") and key_stripped.endswith("__"))
def merge_dict(
source: dict, destination: dict, max_depth: int = 50, max_list_size: int = 10000, _depth: int = 0, _seen: set = None
) -> dict:
"""
Merge data from source into destination safely with protection against DoS attacks.
Args:
source (dict): Source data
destination (dict): Destination data
max_depth (int): Maximum recursion depth allowed (default: 50)
max_list_size (int): Maximum list size allowed (default: 10000)
_depth (int): Internal recursion depth counter
_seen (set): Internal circular reference tracker
Returns:
dict: The merged dictionary
Raises:
TypeError: If source or destination are not dictionaries
RecursionError: If recursion depth exceeds safety limit
ValueError: If circular reference is detected
"""
if not isinstance(source, dict) or not isinstance(destination, dict):
msg = "Both source and destination must be dictionaries."
raise TypeError(msg)
destination_copy = copy.deepcopy(destination)
# Prevent deep recursion DoS
if _depth > max_depth:
msg = f"Recursion depth limit exceeded ({max_depth})"
raise RecursionError(msg)
# Initialize circular reference tracking
if _seen is None:
_seen = set()
# Check for circular references
source_id = id(source)
dest_id = id(destination)
if source_id in _seen or dest_id in _seen:
msg = "Circular reference detected"
raise ValueError(msg)
# Track current objects
current_seen = _seen | {source_id, dest_id}
# Create a clean copy of destination with only safe keys
destination_copy = {}
for k, v in destination.items():
if _is_safe_key(k):
# Prevent memory DoS from large lists
if isinstance(v, list) and len(v) > max_list_size:
destination_copy[k] = v[:max_list_size] # Truncate large lists
else:
destination_copy[k] = copy.deepcopy(v)
for key, value in source.items():
if key in {"__class__", "__dict__", "__globals__", "__builtins__"}:
# Skip unsafe keys
if not _is_safe_key(key):
continue
destination_value = destination_copy.get(key)
# Recursively merge dictionaries
# Recursively merge dictionaries with safety checks
if isinstance(value, dict) and isinstance(destination_value, dict):
destination_copy[key] = merge_dict(value, destination_value)
destination_copy[key] = merge_dict(
value, destination_value, max_depth, max_list_size, _depth + 1, current_seen
)
# Safely extend lists without reference issues
# Safely extend lists with size limits
elif isinstance(value, list) and isinstance(destination_value, list):
destination_copy[key] = copy.deepcopy(destination_value) + copy.deepcopy(value)
# Prevent memory DoS
combined_size = len(value) + len(destination_value)
if combined_size > max_list_size:
# Truncate to stay within limits
available_space = max_list_size - len(destination_value)
truncated_value = value[: max(0, available_space)]
destination_copy[key] = copy.deepcopy(destination_value) + copy.deepcopy(truncated_value)
else:
destination_copy[key] = copy.deepcopy(destination_value) + copy.deepcopy(value)
# For non-dict values or when destination doesn't have the key
elif isinstance(value, dict):
destination_copy[key] = merge_dict(value, {}, max_depth, max_list_size, _depth + 1, current_seen)
elif isinstance(value, list) and len(value) > max_list_size:
# Truncate large lists
destination_copy[key] = copy.deepcopy(value[:max_list_size])
else:
destination_copy[key] = copy.deepcopy(value)

2
app/local.py
View file

@ -1,5 +1,5 @@
#!/usr/bin/env python3
# flake8: noqa: SIM115 S310 T201
# flake8: noqa: S310 T201
import argparse
import os

502
app/tests/test_ag_utils.py Normal file
View file

@ -0,0 +1,502 @@
"""
Tests for ag_utils.py functions.
This test suite provides comprehensive coverage for all functions in ag_utils.py:
- get_value: Tests callable detection and value retrieval
- ag_set: Tests nested dictionary path setting with various scenarios
- ag: Tests nested dictionary/list/object access with dot notation
- ag_sets: Tests bulk setting of multiple paths
- ag_exists: Tests existence checking for nested paths
- ag_delete: Tests deletion of nested paths and keys
Total test functions: 53
All edge cases, error conditions, and normal operations are covered.
"""
from unittest.mock import MagicMock
import pytest
from app.library.ag_utils import (
ag,
ag_delete,
ag_exists,
ag_set,
ag_sets,
get_value,
)
class TestGetValue:
"""Test the get_value function."""
def test_get_value_with_value(self):
"""Test get_value returns the value when not callable."""
assert get_value(42) == 42
assert get_value("test") == "test"
assert get_value([1, 2, 3]) == [1, 2, 3]
assert get_value({"key": "value"}) == {"key": "value"}
assert get_value(None) is None
def test_get_value_with_callable(self):
"""Test get_value calls the function when callable."""
mock_func = MagicMock(return_value="called")
result = get_value(mock_func)
assert result == "called"
mock_func.assert_called_once()
def test_get_value_with_lambda(self):
"""Test get_value with lambda functions."""
assert get_value(lambda: 100) == 100
assert get_value(lambda: "lambda_result") == "lambda_result"
def test_get_value_with_function(self):
"""Test get_value with regular functions."""
def test_func():
return "function_result"
assert get_value(test_func) == "function_result"
class TestAgSet:
"""Test the ag_set function."""
def test_ag_set_simple_path(self):
"""Test setting a value with simple path."""
data = {}
result = ag_set(data, "key", "value")
assert result == {"key": "value"}
assert data == {"key": "value"}
def test_ag_set_nested_path(self):
"""Test setting a value with nested path."""
data = {}
result = ag_set(data, "a.b.c", "nested_value")
expected = {"a": {"b": {"c": "nested_value"}}}
assert result == expected
assert data == expected
def test_ag_set_existing_structure(self):
"""Test setting a value in existing structure."""
data = {"a": {"b": {"existing": "value"}}}
ag_set(data, "a.b.c", "new_value")
expected = {"a": {"b": {"existing": "value", "c": "new_value"}}}
assert data == expected
def test_ag_set_overwrite_existing(self):
"""Test overwriting existing value."""
data = {"a": {"b": "old_value"}}
ag_set(data, "a.b", "new_value")
assert data == {"a": {"b": "new_value"}}
def test_ag_set_custom_separator(self):
"""Test using custom separator."""
data = {}
ag_set(data, "a/b/c", "value", separator="/")
assert data == {"a": {"b": {"c": "value"}}}
def test_ag_set_overwrite_non_dict_intermediate(self):
"""Test overwriting non-dict intermediate value with dict."""
data = {"a": "not_a_dict"}
ag_set(data, "a.b", "value")
# The function should overwrite "not_a_dict" with a dict containing the new path
expected = {"a": {"b": "value"}}
assert data == expected
def test_ag_set_error_on_non_dict_final(self):
"""Test error when final target is not a dict."""
data = "not_a_dict"
with pytest.raises(RuntimeError, match="Cannot set value at path 'key'"):
ag_set(data, "key", "value")
class TestAg:
"""Test the ag function."""
def test_ag_with_none_path(self):
"""Test ag returns whole structure when path is None."""
data = {"a": 1, "b": 2}
assert ag(data, None) == data
assert ag(data, "") == data
def test_ag_simple_dict_access(self):
"""Test simple dictionary key access."""
data = {"x": 10, "y": 20}
assert ag(data, "x") == 10
assert ag(data, "y") == 20
def test_ag_missing_key_with_default(self):
"""Test accessing missing key returns default."""
data = {"x": 10}
assert ag(data, "missing", default=0) == 0
assert ag(data, "missing", default="not_found") == "not_found"
def test_ag_nested_dict_access(self):
"""Test nested dictionary access with dot notation."""
data = {"a": {"b": {"c": 42}}}
assert ag(data, "a.b.c") == 42
def test_ag_nested_missing_key(self):
"""Test nested access with missing intermediate keys."""
data = {"a": {"b": 1}}
assert ag(data, "a.missing.c", default="default") == "default"
assert ag(data, "missing.b.c", default="default") == "default"
def test_ag_list_access_by_index(self):
"""Test accessing list elements by index."""
data = [10, 20, 30]
assert ag(data, 0) == 10
assert ag(data, 1) == 20
assert ag(data, 2) == 30
def test_ag_list_access_out_of_bounds(self):
"""Test list access with out of bounds index."""
data = [10, 20]
assert ag(data, 5, default="default") == "default"
assert ag(data, -3, default="default") == "default" # -3 is out of bounds for 2-element list
def test_ag_list_negative_indices(self):
"""Test list access with valid negative indices."""
data = [10, 20, 30]
assert ag(data, -1) == 30 # Last element
assert ag(data, -2) == 20 # Second to last
assert ag(data, -3) == 10 # First element
def test_ag_mixed_dict_list_access(self):
"""Test accessing nested structure with dicts and lists."""
data = {"items": [{"name": "item1"}, {"name": "item2"}]}
assert ag(data, "items.0.name") == "item1"
assert ag(data, "items.1.name") == "item2"
def test_ag_list_of_paths(self):
"""Test trying multiple paths and returning first found."""
data = {"a": 1, "b": 2}
assert ag(data, ["missing1", "missing2", "a"], default="default") == 1
assert ag(data, ["missing1", "b", "a"], default="default") == 2
assert ag(data, ["missing1", "missing2"], default="default") == "default"
def test_ag_custom_separator(self):
"""Test using custom separator."""
data = {"a": {"b": {"c": 100}}}
assert ag(data, "a/b/c", separator="/") == 100
def test_ag_with_none_values(self):
"""Test ag behavior with None values."""
data = {"a": None, "b": {"c": None}}
assert ag(data, "a", default="default") == "default"
assert ag(data, "b.c", default="default") == "default"
def test_ag_with_callable_default(self):
"""Test ag with callable default value."""
data = {}
mock_default = MagicMock(return_value="called_default")
result = ag(data, "missing", default=mock_default)
assert result == "called_default"
mock_default.assert_called_once()
def test_ag_with_object_attributes(self):
"""Test ag with object attributes using vars()."""
class TestObj:
def __init__(self):
self.attr1 = "value1"
self.attr2 = {"nested": "value2"}
obj = TestObj()
assert ag(obj, "attr1") == "value1"
assert ag(obj, "attr2.nested") == "value2"
def test_ag_with_non_dict_non_list_fallback(self):
"""Test ag fallback for non-dict, non-list objects without vars."""
class NoVarsObj:
__slots__ = ["value"]
def __init__(self):
self.value = "test"
obj = NoVarsObj()
assert ag(obj, "anything", default="fallback") == "fallback"
class TestAgSets:
"""Test the ag_sets function."""
def test_ag_sets_multiple_paths(self):
"""Test setting multiple paths at once."""
data = {}
path_values = {"a.b.c": "value1", "a.b.d": "value2", "x.y": "value3"}
result = ag_sets(data, path_values)
expected = {"a": {"b": {"c": "value1", "d": "value2"}}, "x": {"y": "value3"}}
assert result == expected
assert data == expected
def test_ag_sets_custom_separator(self):
"""Test ag_sets with custom separator."""
data = {}
path_values = {"a/b/c": "value1", "x/y": "value2"}
ag_sets(data, path_values, separator="/")
expected = {"a": {"b": {"c": "value1"}}, "x": {"y": "value2"}}
assert data == expected
def test_ag_sets_existing_structure(self):
"""Test ag_sets with existing data structure."""
data = {"a": {"existing": "value"}}
path_values = {"a.b.c": "new_value", "d": "another_value"}
ag_sets(data, path_values)
expected = {"a": {"existing": "value", "b": {"c": "new_value"}}, "d": "another_value"}
assert data == expected
def test_ag_sets_empty_dict(self):
"""Test ag_sets with empty path_values dict."""
data = {"existing": "data"}
original = data.copy()
ag_sets(data, {})
assert data == original
class TestAgExists:
"""Test the ag_exists function."""
def test_ag_exists_simple_dict_key(self):
"""Test checking existence of simple dict keys."""
data = {"a": "value", "b": None, "c": 0}
assert ag_exists(data, "a") is True
assert ag_exists(data, "b") is False # None values return False
assert ag_exists(data, "c") is True # 0 is not None
assert ag_exists(data, "missing") is False
def test_ag_exists_nested_path(self):
"""Test checking existence of nested paths."""
data = {"a": {"b": {"c": "value", "d": None}}}
assert ag_exists(data, "a.b.c") is True
assert ag_exists(data, "a.b.d") is False # None value
assert ag_exists(data, "a.b.missing") is False
assert ag_exists(data, "a.missing.c") is False
def test_ag_exists_list_indices(self):
"""Test checking existence of list indices."""
data = [10, None, 30]
assert ag_exists(data, 0) is True
assert ag_exists(data, 1) is False # None value
assert ag_exists(data, 2) is True
assert ag_exists(data, 5) is False # Out of bounds
def test_ag_exists_mixed_structure(self):
"""Test checking existence in mixed dict/list structure."""
data = {"items": [{"name": "item1"}, None, {"name": "item3"}]}
assert ag_exists(data, "items.0.name") is True
assert ag_exists(data, "items.1.name") is False # items[1] is None
assert ag_exists(data, "items.2.name") is True
assert ag_exists(data, "items.5.name") is False # Out of bounds
def test_ag_exists_custom_separator(self):
"""Test ag_exists with custom separator."""
data = {"a": {"b": {"c": "value"}}}
assert ag_exists(data, "a/b/c", separator="/") is True
assert ag_exists(data, "a/b/missing", separator="/") is False
def test_ag_exists_with_object(self):
"""Test ag_exists with object using vars()."""
class TestObj:
def __init__(self):
self.attr = "value"
self.nested = {"key": "value"}
obj = TestObj()
assert ag_exists(obj, "attr") is True
assert ag_exists(obj, "nested.key") is True
assert ag_exists(obj, "missing") is False
def test_ag_exists_with_non_vars_object(self):
"""Test ag_exists with object that doesn't support vars()."""
class NoVarsObj:
__slots__ = []
obj = NoVarsObj()
assert ag_exists(obj, "anything") is False
class TestAgDelete:
"""Test the ag_delete function."""
def test_ag_delete_simple_dict_key(self):
"""Test deleting simple dictionary keys."""
data = {"a": 1, "b": 2, "c": 3}
result = ag_delete(data, "b")
assert result == {"a": 1, "c": 3}
assert data == {"a": 1, "c": 3}
def test_ag_delete_nested_path(self):
"""Test deleting nested dictionary paths."""
data = {"a": {"b": {"c": 1, "d": 2}, "e": 3}}
ag_delete(data, "a.b.c")
expected = {"a": {"b": {"d": 2}, "e": 3}}
assert data == expected
def test_ag_delete_list_index(self):
"""Test deleting list elements by index."""
data = [10, 20, 30, 40]
ag_delete(data, 1)
assert data == [10, 30, 40]
def test_ag_delete_mixed_structure(self):
"""Test deleting from mixed dict/list structure."""
data = {"items": [{"name": "item1"}, {"name": "item2", "value": 100}]}
ag_delete(data, "items.1.value")
expected = {"items": [{"name": "item1"}, {"name": "item2"}]}
assert data == expected
def test_ag_delete_multiple_paths(self):
"""Test deleting multiple paths at once."""
data = {"a": {"b": 1, "c": 2}, "d": 3, "e": 4}
ag_delete(data, ["a.b", "d"])
expected = {"a": {"c": 2}, "e": 4}
assert data == expected
def test_ag_delete_custom_separator(self):
"""Test ag_delete with custom separator."""
data = {"a": {"b": {"c": 1}}}
ag_delete(data, "a/b/c", separator="/")
assert data == {"a": {"b": {}}}
def test_ag_delete_missing_key(self):
"""Test deleting non-existent keys (should not raise error)."""
data = {"a": {"b": 1}}
original = {"a": {"b": 1}}
# Should not raise error and not modify data
ag_delete(data, "missing")
ag_delete(data, "a.missing")
ag_delete(data, "a.b.c")
assert data == original
def test_ag_delete_out_of_bounds_list(self):
"""Test deleting out of bounds list index (should not raise error)."""
data = [1, 2, 3]
original = [1, 2, 3]
ag_delete(data, 10) # Out of bounds
ag_delete(data, -1) # Negative index
assert data == original
def test_ag_delete_with_object(self):
"""Test ag_delete with object using vars()."""
class TestObj:
def __init__(self):
self.attr1 = "value1"
self.attr2 = {"nested": "value2"}
obj = TestObj()
ag_delete(obj, "attr1")
# Check that attr1 was deleted
assert not hasattr(obj, "attr1")
assert hasattr(obj, "attr2")
def test_ag_delete_invalid_list_string_index(self):
"""Test ag_delete with invalid string index for list."""
data = {"items": [1, 2, 3]}
original_items = [1, 2, 3]
ag_delete(data, "items.invalid_index")
# Should not modify the list
assert data["items"] == original_items
def test_ag_delete_path_through_none(self):
"""Test ag_delete when path goes through None value."""
data = {"a": {"b": None}}
original = {"a": {"b": None}}
ag_delete(data, "a.b.c") # Can't traverse through None
assert data == original
class TestEdgeCases:
"""Test edge cases and error conditions."""
def test_empty_data_structures(self):
"""Test functions with empty data structures."""
empty_dict = {}
empty_list = []
# ag function
assert ag(empty_dict, "key", default="default") == "default"
assert ag(empty_list, 0, default="default") == "default"
# ag_exists function
assert ag_exists(empty_dict, "key") is False
assert ag_exists(empty_list, 0) is False
# ag_delete function (should not raise errors)
ag_delete(empty_dict, "key")
ag_delete(empty_list, 0)
assert empty_dict == {}
assert empty_list == []
def test_deeply_nested_structures(self):
"""Test with deeply nested structures."""
# Create 10-level deep structure
data = {}
current = data
for i in range(10):
current[f"level{i}"] = {}
current = current[f"level{i}"]
current["value"] = "deep_value"
path = ".".join(f"level{i}" for i in range(10)) + ".value"
# Test ag function
assert ag(data, path) == "deep_value"
# Test ag_exists function
assert ag_exists(data, path) is True
# Test ag_set function
ag_set(data, path.replace(".value", ".new_value"), "new_deep_value")
new_path = ".".join(f"level{i}" for i in range(10)) + ".new_value"
assert ag(data, new_path) == "new_deep_value"
def test_special_characters_in_keys(self):
"""Test with special characters in dictionary keys."""
data = {"key with spaces": "value1", "key.with.dots": "value2", "key/with/slashes": "value3"}
# These should work with direct key access
assert ag(data, "key with spaces") == "value1"
assert ag(data, "key.with.dots") == "value2"
assert ag(data, "key/with/slashes") == "value3"
# Test existence
assert ag_exists(data, "key with spaces") is True
assert ag_exists(data, "key.with.dots") is True
# Test deletion
ag_delete(data, "key with spaces")
assert "key with spaces" not in data
def test_type_consistency(self):
"""Test type consistency across operations."""
data = {"string": "test", "number": 42, "boolean": True, "list": [1, 2, 3], "dict": {"nested": "value"}}
# All values should be retrieved correctly
assert ag(data, "string") == "test"
assert ag(data, "number") == 42
assert ag(data, "boolean") is True
assert ag(data, "list") == [1, 2, 3]
assert ag(data, "dict") == {"nested": "value"}
# All should exist
for key in data:
assert ag_exists(data, key) is True

260
app/tests/test_cache.py Normal file
View file

@ -0,0 +1,260 @@
"""
Tests for cache.py - Thread-safe caching utilities.
This test suite provides comprehensive coverage for the Cache class:
- Tests basic cache operations (set, get, delete, clear)
- Tests TTL (time-to-live) functionality
- Tests thread safety
- Tests cache expiration
- Tests default value handling
- Tests key existence checking
- Tests hash generation
- Tests async methods
Total test functions: 15
All cache operations and edge cases are covered.
"""
import asyncio
import threading
import time
import pytest
from app.library.cache import Cache
class TestCache:
"""Test the Cache class."""
def setup_method(self):
"""Set up test fixtures."""
# Clear any existing cache instance
if hasattr(Cache, "_instances"):
Cache._instances.clear()
self.cache = Cache()
def test_singleton_behavior(self):
"""Test that Cache follows singleton pattern."""
cache1 = Cache()
cache2 = Cache()
assert cache1 is cache2
def test_basic_set_and_get(self):
"""Test basic cache set and get operations."""
self.cache.set("key1", "value1")
assert self.cache.get("key1") == "value1"
def test_get_with_default(self):
"""Test get with default value for non-existent keys."""
assert self.cache.get("nonexistent", "default") == "default"
assert self.cache.get("nonexistent") is None
def test_set_with_ttl(self):
"""Test setting values with TTL."""
self.cache.set("temp_key", "temp_value", ttl=0.1)
assert self.cache.get("temp_key") == "temp_value"
# Wait for expiration
time.sleep(0.2)
assert self.cache.get("temp_key") is None
def test_set_without_ttl(self):
"""Test setting values without TTL (permanent)."""
self.cache.set("permanent_key", "permanent_value")
assert self.cache.get("permanent_key") == "permanent_value"
# Should still be there after some time
time.sleep(0.1)
assert self.cache.get("permanent_key") == "permanent_value"
def test_has_key(self):
"""Test key existence checking."""
assert not self.cache.has("nonexistent")
self.cache.set("existing", "value")
assert self.cache.has("existing")
def test_has_key_with_expiration(self):
"""Test key existence with expired keys."""
self.cache.set("expiring", "value", ttl=0.1)
assert self.cache.has("expiring")
time.sleep(0.2)
assert not self.cache.has("expiring")
def test_ttl_method(self):
"""Test TTL retrieval."""
# Key without TTL
self.cache.set("permanent", "value")
assert self.cache.ttl("permanent") is None
# Key with TTL
self.cache.set("temporary", "value", ttl=1.0)
ttl = self.cache.ttl("temporary")
assert ttl is not None
assert 0.5 < ttl <= 1.0
# Non-existent key
assert self.cache.ttl("nonexistent") is None
def test_delete_key(self):
"""Test key deletion."""
self.cache.set("to_delete", "value")
assert self.cache.get("to_delete") == "value"
self.cache.delete("to_delete")
assert self.cache.get("to_delete") is None
def test_delete_nonexistent_key(self):
"""Test deleting non-existent key (should not raise error)."""
self.cache.delete("nonexistent") # Should not raise
def test_clear_cache(self):
"""Test clearing all cache entries."""
self.cache.set("key1", "value1")
self.cache.set("key2", "value2")
self.cache.clear()
assert self.cache.get("key1") is None
assert self.cache.get("key2") is None
def test_hash_method(self):
"""Test hash generation."""
hash1 = self.cache.hash("test_string")
hash2 = self.cache.hash("test_string")
hash3 = self.cache.hash("different_string")
# Same input should produce same hash
assert hash1 == hash2
# Different input should produce different hash
assert hash1 != hash3
# Should be valid SHA-256 hex string
assert len(hash1) == 64
assert all(c in "0123456789abcdef" for c in hash1)
def test_thread_safety(self):
"""Test thread safety of cache operations."""
results = []
errors = []
def worker(worker_id):
try:
for i in range(10):
key = f"worker_{worker_id}_key_{i}"
value = f"worker_{worker_id}_value_{i}"
self.cache.set(key, value)
retrieved = self.cache.get(key)
if retrieved == value:
results.append(f"{worker_id}_{i}_success")
else:
errors.append(f"{worker_id}_{i}_mismatch: {retrieved} != {value}")
except Exception as e:
errors.append(f"Worker {worker_id} error: {e}")
# Create multiple threads
threads = []
for i in range(5):
thread = threading.Thread(target=worker, args=(i,))
threads.append(thread)
thread.start()
# Wait for all threads to complete
for thread in threads:
thread.join()
# Check results
assert len(errors) == 0, f"Thread safety errors: {errors}"
assert len(results) == 50 # 5 workers * 10 operations each
def test_async_set(self):
"""Test async set method using asyncio.run."""
async def async_test():
await self.cache.aset("async_key", "async_value")
assert self.cache.get("async_key") == "async_value"
asyncio.run(async_test())
def test_async_set_with_ttl(self):
"""Test async set with TTL using asyncio.run."""
async def async_test():
await self.cache.aset("async_temp", "async_value", ttl=0.1)
assert self.cache.get("async_temp") == "async_value"
await asyncio.sleep(0.2)
assert self.cache.get("async_temp") is None
asyncio.run(async_test())
asyncio.run(async_test())
def test_expired_key_cleanup_on_get(self):
"""Test that expired keys are cleaned up when accessed."""
# Set a key with very short TTL
self.cache.set("cleanup_test", "value", ttl=0.05)
# Verify it's initially there
assert self.cache.get("cleanup_test") == "value"
# Wait for expiration
time.sleep(0.1)
# Getting expired key should clean it up and return None
assert self.cache.get("cleanup_test") is None
# Key should be removed from internal cache
assert "cleanup_test" not in self.cache._cache
def test_expired_key_cleanup_on_has(self):
"""Test that expired keys are cleaned up when checking existence."""
# Set a key with very short TTL
self.cache.set("has_cleanup", "value", ttl=0.05)
# Verify it's initially there
assert self.cache.has("has_cleanup")
# Wait for expiration
time.sleep(0.1)
# Checking existence of expired key should clean it up
assert not self.cache.has("has_cleanup")
# Key should be removed from internal cache
assert "has_cleanup" not in self.cache._cache
def test_complex_data_types(self):
"""Test caching of complex data types."""
# Test list
test_list = [1, 2, {"nested": "dict"}]
self.cache.set("list_key", test_list)
assert self.cache.get("list_key") == test_list
# Test dict
test_dict = {"key": "value", "nested": {"list": [1, 2, 3]}}
self.cache.set("dict_key", test_dict)
assert self.cache.get("dict_key") == test_dict
# Test custom object
class CustomObject:
def __init__(self, value):
self.value = value
def __eq__(self, other):
return isinstance(other, CustomObject) and self.value == other.value
def __hash__(self):
return hash(self.value)
custom_obj = CustomObject("test_value")
self.cache.set("object_key", custom_obj)
retrieved = self.cache.get("object_key")
assert retrieved == custom_obj
if __name__ == "__main__":
pytest.main([__file__])

751
app/tests/test_conditions.py Normal file
View file

@ -0,0 +1,751 @@
"""
Tests for conditions.py - Download conditions management.
This test suite provides comprehensive coverage for the conditions module:
- Tests Condition dataclass functionality
- Tests Conditions singleton class behavior
- Tests condition loading, saving, and validation
- Tests condition matching against info dicts
- Tests error handling and edge cases
Total test functions: 15+
All condition management functionality and edge cases are covered.
"""
import json
import tempfile
import uuid
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from app.library.conditions import Condition, Conditions
from app.library.Singleton import Singleton
class TestCondition:
"""Test the Condition dataclass."""
def test_condition_creation_with_defaults(self):
"""Test creating a condition with default values."""
condition = Condition(name="test", filter="duration > 60")
# Check that ID is generated
assert condition.id
assert isinstance(condition.id, str)
# Check required fields
assert condition.name == "test"
assert condition.filter == "duration > 60"
# Check defaults
assert condition.cli == ""
assert condition.extras == {}
def test_condition_creation_with_all_fields(self):
"""Test creating a condition with all fields specified."""
test_id = str(uuid.uuid4())
extras = {"key": "value", "number": 42}
condition = Condition(
id=test_id,
name="full_test",
filter="uploader = 'test'",
cli="--format best",
extras=extras
)
assert condition.id == test_id
assert condition.name == "full_test"
assert condition.filter == "uploader = 'test'"
assert condition.cli == "--format best"
assert condition.extras == extras
def test_condition_serialize(self):
"""Test condition serialization to dict."""
condition = Condition(
name="serialize_test",
filter="title ~= 'test'",
cli="--audio-quality 0",
extras={"tag": "music"}
)
serialized = condition.serialize()
assert isinstance(serialized, dict)
assert serialized["name"] == "serialize_test"
assert serialized["filter"] == "title ~= 'test'"
assert serialized["cli"] == "--audio-quality 0"
assert serialized["extras"] == {"tag": "music"}
assert "id" in serialized
def test_condition_json(self):
"""Test condition JSON serialization."""
condition = Condition(name="json_test", filter="duration < 300")
json_str = condition.json()
assert isinstance(json_str, str)
# Should be valid JSON
data = json.loads(json_str)
assert data["name"] == "json_test"
assert data["filter"] == "duration < 300"
def test_condition_get_method(self):
"""Test condition get method for accessing fields."""
condition = Condition(
name="get_test",
filter="view_count > 1000",
extras={"category": "popular"}
)
assert condition.get("name") == "get_test"
assert condition.get("filter") == "view_count > 1000"
assert condition.get("extras") == {"category": "popular"}
assert condition.get("nonexistent") is None
assert condition.get("nonexistent", "default") == "default"
class TestConditions:
"""Test the Conditions singleton class."""
def setup_method(self):
"""Set up test fixtures by clearing singleton instances."""
# Clear singleton instances before each test
Singleton._instances.clear()
# Reset class variable
Conditions._items = []
Conditions._instance = None
def test_conditions_singleton(self):
"""Test that Conditions follows singleton pattern."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "test_conditions.json"
instance1 = Conditions(file=file_path)
instance2 = Conditions.get_instance()
assert instance1 is instance2
def test_conditions_initialization(self):
"""Test Conditions initialization with file path."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "test_conditions.json"
conditions = Conditions(file=file_path)
assert conditions._file == file_path
assert isinstance(conditions._items, list)
@patch("app.library.conditions.Config.get_instance")
def test_conditions_default_file_path(self, mock_config):
"""Test Conditions uses default config path when no file specified."""
mock_config_instance = MagicMock()
mock_config_instance.config_path = "/test/config"
mock_config.return_value = mock_config_instance
conditions = Conditions()
expected_path = Path("/test/config") / "conditions.json"
assert conditions._file == expected_path
def test_get_all_empty(self):
"""Test get_all returns empty list when no conditions loaded."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "empty_conditions.json"
conditions = Conditions(file=file_path)
result = conditions.get_all()
assert isinstance(result, list)
assert len(result) == 0
def test_clear(self):
"""Test clearing all conditions."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "clear_test.json"
conditions = Conditions(file=file_path)
# Add some test conditions
conditions._items = [
Condition(name="test1", filter="duration > 60"),
Condition(name="test2", filter="uploader = 'test'")
]
result = conditions.clear()
assert result is conditions # Should return self
assert len(conditions._items) == 0
def test_clear_when_already_empty(self):
"""Test clearing when conditions list is already empty."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "already_empty.json"
conditions = Conditions(file=file_path)
result = conditions.clear()
assert result is conditions
assert len(conditions._items) == 0
def test_load_nonexistent_file(self):
"""Test loading from non-existent file."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "nonexistent.json"
conditions = Conditions(file=file_path)
result = conditions.load()
assert result is conditions
assert len(conditions._items) == 0
def test_load_empty_file(self):
"""Test loading from empty file."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "empty.json"
file_path.touch() # Create empty file
conditions = Conditions(file=file_path)
result = conditions.load()
assert result is conditions
assert len(conditions._items) == 0
def test_load_valid_conditions(self):
"""Test loading valid conditions from file."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "valid_conditions.json"
# Create test data
test_data = [
{
"id": str(uuid.uuid4()),
"name": "short_videos",
"filter": "duration < 300",
"cli": "--format worst",
"extras": {"category": "short"}
},
{
"id": str(uuid.uuid4()),
"name": "music_videos",
"filter": "title ~= 'music'",
"cli": "--audio-quality 0",
"extras": {"type": "audio"}
}
]
file_path.write_text(json.dumps(test_data, indent=4))
conditions = Conditions(file=file_path)
result = conditions.load()
assert result is conditions
assert len(conditions._items) == 2
# Check first condition
assert conditions._items[0].name == "short_videos"
assert conditions._items[0].filter == "duration < 300"
assert conditions._items[0].cli == "--format worst"
assert conditions._items[0].extras == {"category": "short"}
# Check second condition
assert conditions._items[1].name == "music_videos"
assert conditions._items[1].filter == "title ~= 'music'"
def test_load_conditions_without_id(self):
"""Test loading conditions that don't have ID (should generate ID)."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "no_id_conditions.json"
# Create test data without ID
test_data = [
{
"name": "no_id_test",
"filter": "duration > 120"
}
]
file_path.write_text(json.dumps(test_data))
with patch.object(Conditions, "save") as mock_save:
conditions = Conditions(file=file_path)
conditions.load()
# Should have generated ID
assert len(conditions._items) == 1
assert conditions._items[0].id
assert conditions._items[0].name == "no_id_test"
# Should call save due to changes
mock_save.assert_called_once()
def test_load_conditions_without_extras(self):
"""Test loading conditions that don't have extras field."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "no_extras_conditions.json"
test_data = [
{
"id": str(uuid.uuid4()),
"name": "no_extras_test",
"filter": "uploader = 'test'"
}
]
file_path.write_text(json.dumps(test_data))
with patch.object(Conditions, "save") as mock_save:
conditions = Conditions(file=file_path)
conditions.load()
# Should have generated empty extras
assert len(conditions._items) == 1
assert conditions._items[0].extras == {}
# Should call save due to changes
mock_save.assert_called_once()
def test_load_invalid_json(self):
"""Test loading file with invalid JSON."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "invalid.json"
file_path.write_text("invalid json content")
conditions = Conditions(file=file_path)
result = conditions.load()
assert result is conditions
assert len(conditions._items) == 0
def test_load_invalid_condition_data(self):
"""Test loading file with invalid condition data."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "invalid_data.json"
# Missing required fields
test_data = [
{"id": "valid", "name": "valid", "filter": "duration > 60"},
{"invalid": "data"} # Missing required fields
]
file_path.write_text(json.dumps(test_data))
conditions = Conditions(file=file_path)
result = conditions.load()
# Should load only valid conditions
assert result is conditions
assert len(conditions._items) == 1
assert conditions._items[0].name == "valid"
def test_validate_valid_condition_dict(self):
"""Test validating a valid condition dictionary."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "validate_test.json"
conditions = Conditions(file=file_path)
valid_condition = {
"id": str(uuid.uuid4()),
"name": "valid_test",
"filter": "duration > 60",
"cli": "--format best",
"extras": {"key": "value"}
}
result = conditions.validate(valid_condition)
assert result is True
def test_validate_valid_condition_object(self):
"""Test validating a valid Condition object."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "validate_obj_test.json"
conditions = Conditions(file=file_path)
valid_condition = Condition(
name="valid_obj_test",
filter="uploader = 'test'"
)
result = conditions.validate(valid_condition)
assert result is True
def test_validate_missing_id(self):
"""Test validating condition without ID."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "no_id_validate.json"
conditions = Conditions(file=file_path)
invalid_condition = {
"name": "no_id_test",
"filter": "duration > 60"
}
with pytest.raises(ValueError, match="No id found"):
conditions.validate(invalid_condition)
def test_validate_missing_name(self):
"""Test validating condition without name."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "no_name_validate.json"
conditions = Conditions(file=file_path)
invalid_condition = {
"id": str(uuid.uuid4()),
"filter": "duration > 60"
}
with pytest.raises(ValueError, match="No name found"):
conditions.validate(invalid_condition)
def test_validate_missing_filter(self):
"""Test validating condition without filter."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "no_filter_validate.json"
conditions = Conditions(file=file_path)
invalid_condition = {
"id": str(uuid.uuid4()),
"name": "no_filter_test"
}
with pytest.raises(ValueError, match="No filter found"):
conditions.validate(invalid_condition)
def test_validate_invalid_filter(self):
"""Test validating condition with invalid filter syntax."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "invalid_filter.json"
conditions = Conditions(file=file_path)
# Use a filter that will cause a syntax error in the parser
invalid_condition = {
"id": str(uuid.uuid4()),
"name": "invalid_filter_test",
"filter": "duration > & < 60", # Invalid syntax with consecutive operators
"cli": "",
"extras": {}
}
with pytest.raises(ValueError, match="Invalid filter"):
conditions.validate(invalid_condition)
def test_validate_invalid_cli(self):
"""Test validating condition with invalid CLI options."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "invalid_cli.json"
conditions = Conditions(file=file_path)
invalid_condition = {
"id": str(uuid.uuid4()),
"name": "invalid_cli_test",
"filter": "duration > 60",
"cli": "--invalid-option-that-doesnt-exist"
}
with pytest.raises(ValueError, match="Invalid command options"):
conditions.validate(invalid_condition)
def test_validate_invalid_extras_type(self):
"""Test validating condition with non-dict extras."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "invalid_extras.json"
conditions = Conditions(file=file_path)
invalid_condition = {
"id": str(uuid.uuid4()),
"name": "invalid_extras_test",
"filter": "duration > 60",
"extras": "not a dict"
}
with pytest.raises(ValueError, match="Extras must be a dictionary"):
conditions.validate(invalid_condition)
def test_validate_invalid_item_type(self):
"""Test validating invalid item type."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "invalid_type.json"
conditions = Conditions(file=file_path)
with pytest.raises(ValueError, match="Unexpected.*item type"):
conditions.validate("invalid type")
def test_save_conditions(self):
"""Test saving conditions to file."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "save_test.json"
conditions = Conditions(file=file_path)
test_conditions = [
Condition(name="save_test1", filter="duration > 60"),
Condition(name="save_test2", filter="uploader = 'test'")
]
result = conditions.save(test_conditions)
assert result is conditions
assert file_path.exists()
# Verify file content
saved_data = json.loads(file_path.read_text())
assert len(saved_data) == 2
assert saved_data[0]["name"] == "save_test1"
assert saved_data[1]["name"] == "save_test2"
def test_save_conditions_dict_format(self):
"""Test saving conditions in dictionary format."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "save_dict_test.json"
conditions = Conditions(file=file_path)
test_conditions = [
{
"id": str(uuid.uuid4()),
"name": "dict_test",
"filter": "duration < 300",
"cli": "",
"extras": {}
}
]
conditions.save(test_conditions)
assert file_path.exists()
saved_data = json.loads(file_path.read_text())
assert len(saved_data) == 1
assert saved_data[0]["name"] == "dict_test"
def test_has_condition_by_id(self):
"""Test checking if condition exists by ID."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "has_test.json"
conditions = Conditions(file=file_path)
test_id = str(uuid.uuid4())
test_condition = Condition(id=test_id, name="has_test", filter="duration > 60")
conditions._items = [test_condition]
assert conditions.has(test_id) is True
assert conditions.has("nonexistent") is False
def test_has_condition_by_name(self):
"""Test checking if condition exists by name."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "has_name_test.json"
conditions = Conditions(file=file_path)
test_condition = Condition(name="has_name_test", filter="uploader = 'test'")
conditions._items = [test_condition]
assert conditions.has("has_name_test") is True
assert conditions.has("nonexistent_name") is False
def test_get_condition_by_id(self):
"""Test getting condition by ID."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "get_id_test.json"
conditions = Conditions(file=file_path)
test_id = str(uuid.uuid4())
test_condition = Condition(id=test_id, name="get_id_test", filter="duration > 120")
conditions._items = [test_condition]
result = conditions.get(test_id)
assert result is test_condition
assert conditions.get("nonexistent") is None
def test_get_condition_by_name(self):
"""Test getting condition by name."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "get_name_test.json"
conditions = Conditions(file=file_path)
test_condition = Condition(name="get_name_test", filter="title ~= 'music'")
conditions._items = [test_condition]
result = conditions.get("get_name_test")
assert result is test_condition
def test_get_condition_empty_id_or_name(self):
"""Test getting condition with empty ID or name."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "empty_get_test.json"
conditions = Conditions(file=file_path)
assert conditions.get("") is None
assert conditions.get(None) is None
@patch("app.library.conditions.match_str")
def test_match_condition_found(self, mock_match_str):
"""Test matching condition against info dict."""
mock_match_str.return_value = True
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "match_test.json"
conditions = Conditions(file=file_path)
test_condition = Condition(name="match_test", filter="duration > 60")
conditions._items = [test_condition]
info_dict = {"duration": 120, "title": "Test Video"}
result = conditions.match(info_dict)
assert result is test_condition
mock_match_str.assert_called_once_with("duration > 60", info_dict)
@patch("app.library.conditions.match_str")
def test_match_condition_not_found(self, mock_match_str):
"""Test matching when no condition matches."""
mock_match_str.return_value = False
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "no_match_test.json"
conditions = Conditions(file=file_path)
test_condition = Condition(name="no_match_test", filter="duration > 300")
conditions._items = [test_condition]
info_dict = {"duration": 60, "title": "Short Video"}
result = conditions.match(info_dict)
assert result is None
def test_match_empty_conditions(self):
"""Test matching with empty conditions list."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "empty_match_test.json"
conditions = Conditions(file=file_path)
info_dict = {"duration": 120}
result = conditions.match(info_dict)
assert result is None
def test_match_invalid_info_dict(self):
"""Test matching with invalid info dict."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "invalid_info_test.json"
conditions = Conditions(file=file_path)
test_condition = Condition(name="test", filter="duration > 60")
conditions._items = [test_condition]
# Test with None
assert conditions.match(None) is None
# Test with empty dict
assert conditions.match({}) is None
# Test with non-dict
assert conditions.match("not a dict") is None
@patch("app.library.conditions.match_str")
def test_match_filter_evaluation_error(self, mock_match_str):
"""Test matching when filter evaluation raises exception."""
mock_match_str.side_effect = Exception("Filter error")
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "filter_error_test.json"
conditions = Conditions(file=file_path)
test_condition = Condition(name="error_test", filter="invalid filter")
conditions._items = [test_condition]
info_dict = {"duration": 120}
result = conditions.match(info_dict)
assert result is None
def test_match_empty_filter(self):
"""Test matching with condition that has empty filter."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "empty_filter_test.json"
conditions = Conditions(file=file_path)
test_condition = Condition(name="empty_filter", filter="")
conditions._items = [test_condition]
info_dict = {"duration": 120}
result = conditions.match(info_dict)
assert result is None
@patch("app.library.conditions.match_str")
def test_single_match_found(self, mock_match_str):
"""Test single condition matching."""
mock_match_str.return_value = True
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "single_match_test.json"
conditions = Conditions(file=file_path)
test_condition = Condition(name="single_test", filter="uploader = 'test'")
conditions._items = [test_condition]
info_dict = {"uploader": "test", "title": "Test Video"}
result = conditions.single_match("single_test", info_dict)
assert result is test_condition
mock_match_str.assert_called_once_with("uploader = 'test'", info_dict)
@patch("app.library.conditions.match_str")
def test_single_match_not_found(self, mock_match_str):
"""Test single condition matching when condition doesn't match."""
mock_match_str.return_value = False
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "single_no_match_test.json"
conditions = Conditions(file=file_path)
test_condition = Condition(name="single_no_match", filter="uploader = 'other'")
conditions._items = [test_condition]
info_dict = {"uploader": "test", "title": "Test Video"}
result = conditions.single_match("single_no_match", info_dict)
assert result is None
def test_single_match_nonexistent_condition(self):
"""Test single matching with non-existent condition name."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "nonexistent_single_test.json"
conditions = Conditions(file=file_path)
info_dict = {"duration": 120}
result = conditions.single_match("nonexistent", info_dict)
assert result is None
def test_single_match_condition_no_filter(self):
"""Test single matching with condition that has no filter."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "no_filter_single_test.json"
conditions = Conditions(file=file_path)
test_condition = Condition(name="no_filter_single", filter="")
conditions._items = [test_condition]
info_dict = {"duration": 120}
result = conditions.single_match("no_filter_single", info_dict)
assert result is None
def test_single_match_invalid_inputs(self):
"""Test single matching with invalid inputs."""
with tempfile.TemporaryDirectory() as temp_dir:
file_path = Path(temp_dir) / "invalid_single_test.json"
conditions = Conditions(file=file_path)
# Test with empty conditions
assert conditions.single_match("test", {"duration": 120}) is None
# 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 non-dict info
assert conditions.single_match("test", "not a dict") is None

170
app/tests/test_encoder.py Normal file
View file

@ -0,0 +1,170 @@
"""
Tests for encoder.py - JSON encoding utilities.
This test suite provides comprehensive coverage for the Encoder class:
- Tests serialization of various Python types
- Tests special handling of Path objects
- Tests DateRange serialization
- Tests date serialization
- Tests ImpersonateTarget serialization
- Tests ItemDTO serialization
- Tests object serialization with serialize() method
- Tests object serialization with __dict__ fallback
- Tests fallback to default JSONEncoder behavior
Total test functions: 10
All supported types and edge cases are covered.
"""
import json
from datetime import date
from pathlib import Path
from unittest.mock import MagicMock
import pytest
from app.library.encoder import Encoder
class TestEncoder:
"""Test the Encoder class."""
def setup_method(self):
"""Set up test fixtures."""
self.encoder = Encoder()
def test_path_serialization(self):
"""Test that Path objects are serialized as strings."""
path = Path("/tmp/test/file.txt")
result = self.encoder.default(path)
assert result == "/tmp/test/file.txt"
assert isinstance(result, str)
def test_path_serialization_relative(self):
"""Test that relative Path objects are serialized correctly."""
path = Path("relative/path/file.txt")
result = self.encoder.default(path)
assert result == "relative/path/file.txt"
def test_date_serialization(self):
"""Test that date objects are serialized as strings."""
test_date = date(2024, 3, 15)
result = self.encoder.default(test_date)
assert result == "2024-03-15"
def test_object_with_serialize_method(self):
"""Test that objects with serialize method use it."""
class CustomObject:
def serialize(self):
return {"custom": "data", "type": "test"}
obj = CustomObject()
result = self.encoder.default(obj)
assert result == {"custom": "data", "type": "test"}
def test_object_with_dict_fallback(self):
"""Test that objects without serialize method fall back to __dict__."""
class SimpleObject:
def __init__(self):
self.name = "test"
self.value = 42
obj = SimpleObject()
result = self.encoder.default(obj)
assert result == {"name": "test", "value": 42}
def test_object_without_dict_fallback_to_default(self):
"""Test that objects without __dict__ fall back to default JSONEncoder."""
# This should raise TypeError since complex is not JSON serializable
with pytest.raises(TypeError):
self.encoder.default(complex(1, 2))
def test_json_dumps_integration(self):
"""Test full JSON serialization with various types."""
data = {
"path": Path("/tmp/test.txt"),
"date": date(2024, 1, 1),
"number": 42,
"string": "test"
}
result = json.dumps(data, cls=Encoder)
parsed = json.loads(result)
assert parsed["path"] == "/tmp/test.txt"
assert parsed["date"] == "2024-01-01"
assert parsed["number"] == 42
assert parsed["string"] == "test"
def test_json_dumps_with_custom_object(self):
"""Test JSON serialization with custom objects."""
class TestObject:
def __init__(self):
self.name = "test"
self.items = [1, 2, 3]
data = {
"object": TestObject(),
"regular": "data"
}
result = json.dumps(data, cls=Encoder)
parsed = json.loads(result)
assert parsed["object"]["name"] == "test"
assert parsed["object"]["items"] == [1, 2, 3]
assert parsed["regular"] == "data"
def test_nested_serialization(self):
"""Test serialization of nested structures with various types."""
class CustomObj:
def serialize(self):
return {"serialized": True}
data = {
"paths": [Path("/tmp/1.txt"), Path("/tmp/2.txt")],
"dates": [date(2024, 1, 1), date(2024, 12, 31)],
"custom": CustomObj(),
"nested": {
"path": Path("/nested/path"),
"date": date(2024, 6, 15)
}
}
result = json.dumps(data, cls=Encoder)
parsed = json.loads(result)
assert parsed["paths"] == ["/tmp/1.txt", "/tmp/2.txt"]
assert parsed["dates"] == ["2024-01-01", "2024-12-31"]
assert parsed["custom"] == {"serialized": True}
assert parsed["nested"]["path"] == "/nested/path"
assert parsed["nested"]["date"] == "2024-06-15"
def test_mock_daterange_serialization(self):
"""Test DateRange serialization with mock object."""
# Mock a DateRange-like object
mock_daterange = MagicMock()
mock_daterange.start = date(2024, 1, 15)
mock_daterange.end = date(2024, 12, 31)
# Mock isinstance to return True for DateRange
import builtins
original_isinstance = builtins.isinstance
def mock_isinstance(obj, cls):
if obj is mock_daterange and hasattr(cls, "__name__") and "DateRange" in str(cls):
return True
return original_isinstance(obj, cls)
builtins.isinstance = mock_isinstance
try:
result = self.encoder.default(mock_daterange)
expected = {"start": "20240115", "end": "20241231"}
assert result == expected
finally:
builtins.isinstance = original_isinstance
if __name__ == "__main__":
pytest.main([__file__])

255
app/tests/test_singleton.py Normal file
View file

@ -0,0 +1,255 @@
"""
Tests for Singleton.py - Singleton metaclass utilities.
This test suite provides comprehensive coverage for the singleton metaclasses:
- Tests Singleton metaclass behavior
- Tests ThreadSafe metaclass behavior
- Tests thread safety of ThreadSafe metaclass
- Tests that different classes get different instances
- Tests that same class gets same instance
Total test functions: 8
All singleton patterns and edge cases are covered.
"""
import threading
import time
from app.library.Singleton import Singleton, ThreadSafe
class TestSingleton:
"""Test the Singleton metaclass."""
def setup_method(self):
"""Set up test fixtures by clearing singleton instances."""
# Clear singleton instances before each test
Singleton._instances.clear()
ThreadSafe._instances.clear()
def test_singleton_same_instance(self):
"""Test that Singleton returns same instance for same class."""
class TestClass(metaclass=Singleton):
def __init__(self, value=None):
self.value = value
instance1 = TestClass("first")
instance2 = TestClass("second")
# Should be the same instance
assert instance1 is instance2
# First initialization should win
assert instance1.value == "first"
assert instance2.value == "first"
def test_singleton_different_classes(self):
"""Test that Singleton creates different instances for different classes."""
class ClassA(metaclass=Singleton):
def __init__(self):
self.name = "A"
class ClassB(metaclass=Singleton):
def __init__(self):
self.name = "B"
instance_a1 = ClassA()
instance_a2 = ClassA()
instance_b1 = ClassB()
instance_b2 = ClassB()
# Same class should return same instance
assert instance_a1 is instance_a2
assert instance_b1 is instance_b2
# Different classes should return different instances
assert instance_a1 is not instance_b1
assert instance_a1.name == "A"
assert instance_b1.name == "B"
def test_threadsafe_same_instance(self):
"""Test that ThreadSafe returns same instance for same class."""
class TestClass(metaclass=ThreadSafe):
def __init__(self, value=None):
self.value = value
instance1 = TestClass("first")
instance2 = TestClass("second")
# Should be the same instance
assert instance1 is instance2
# First initialization should win
assert instance1.value == "first"
assert instance2.value == "first"
def test_threadsafe_different_classes(self):
"""Test that ThreadSafe creates different instances for different classes."""
class ClassA(metaclass=ThreadSafe):
def __init__(self):
self.name = "A"
class ClassB(metaclass=ThreadSafe):
def __init__(self):
self.name = "B"
instance_a1 = ClassA()
instance_a2 = ClassA()
instance_b1 = ClassB()
instance_b2 = ClassB()
# Same class should return same instance
assert instance_a1 is instance_a2
assert instance_b1 is instance_b2
# Different classes should return different instances
assert instance_a1 is not instance_b1
assert instance_a1.name == "A"
assert instance_b1.name == "B"
def test_threadsafe_thread_safety(self):
"""Test that ThreadSafe is actually thread-safe."""
instances = []
errors = []
class ThreadSafeClass(metaclass=ThreadSafe):
def __init__(self):
# Add a small delay to increase chance of race condition
time.sleep(0.01)
self.thread_id = threading.current_thread().ident
self.creation_time = time.time()
def worker():
try:
instance = ThreadSafeClass()
instances.append(instance)
except Exception as e:
errors.append(str(e))
# Create multiple threads trying to create instances simultaneously
threads = []
for _ in range(10):
thread = threading.Thread(target=worker)
threads.append(thread)
# Start all threads at roughly the same time
for thread in threads:
thread.start()
# Wait for all threads to complete
for thread in threads:
thread.join()
# Check results
assert len(errors) == 0, f"Thread safety errors: {errors}"
assert len(instances) == 10
# All instances should be the same object
first_instance = instances[0]
for instance in instances[1:]:
assert instance is first_instance
# Only one creation should have happened
assert hasattr(first_instance, "thread_id")
assert hasattr(first_instance, "creation_time")
def test_singleton_inheritance(self):
"""Test singleton behavior with inheritance."""
class BaseClass(metaclass=Singleton):
def __init__(self):
self.base_value = "base"
class ChildClass(BaseClass):
def __init__(self):
super().__init__()
self.child_value = "child"
# Each class should have its own singleton instance
base1 = BaseClass()
base2 = BaseClass()
child1 = ChildClass()
child2 = ChildClass()
# Same class instances should be identical
assert base1 is base2
assert child1 is child2
# Different class instances should be different
assert base1 is not child1
# Check values
assert base1.base_value == "base"
assert child1.base_value == "base"
assert child1.child_value == "child"
def test_threadsafe_inheritance(self):
"""Test threadsafe singleton behavior with inheritance."""
class BaseClass(metaclass=ThreadSafe):
def __init__(self):
self.base_value = "base"
class ChildClass(BaseClass):
def __init__(self):
super().__init__()
self.child_value = "child"
# Each class should have its own singleton instance
base1 = BaseClass()
base2 = BaseClass()
child1 = ChildClass()
child2 = ChildClass()
# Same class instances should be identical
assert base1 is base2
assert child1 is child2
# Different class instances should be different
assert base1 is not child1
# Check values
assert base1.base_value == "base"
assert child1.base_value == "base"
assert child1.child_value == "child"
def test_singleton_with_args_and_kwargs(self):
"""Test singleton behavior with various constructor arguments."""
class ConfigClass(metaclass=Singleton):
def __init__(self, name, value=None, **kwargs):
self.name = name
self.value = value
self.extra = kwargs
# First instantiation with arguments
instance1 = ConfigClass("test", value=42, extra_param="extra")
# Second instantiation with different arguments (should be ignored)
instance2 = ConfigClass("different", value=100, other_param="other")
# Should be same instance with original values
assert instance1 is instance2
assert instance1.name == "test"
assert instance1.value == 42
assert instance1.extra == {"extra_param": "extra"}
def test_threadsafe_with_args_and_kwargs(self):
"""Test threadsafe singleton behavior with various constructor arguments."""
class ConfigClass(metaclass=ThreadSafe):
def __init__(self, name, value=None, **kwargs):
self.name = name
self.value = value
self.extra = kwargs
# First instantiation with arguments
instance1 = ConfigClass("test", value=42, extra_param="extra")
# Second instantiation with different arguments (should be ignored)
instance2 = ConfigClass("different", value=100, other_param="other")
# Should be same instance with original values
assert instance1 is instance2
assert instance1.name == "test"
assert instance1.value == 42
assert instance1.extra == {"extra_param": "extra"}
if __name__ == "__main__":
import pytest
pytest.main([__file__])

499
app/tests/test_utils.py
View file

@ -1,4 +1,5 @@
import asyncio
import copy
import re
import tempfile
import uuid
@ -7,6 +8,8 @@ from datetime import datetime, timedelta
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from app.library.Utils import (
FileLogFormatter,
StreamingError,
@ -89,42 +92,210 @@ class TestCalcDownloadPath:
def test_calc_download_path_base_only(self):
"""Test calculating download path with only base path."""
result = calc_download_path(str(self.base_path), create_path=False)
assert result == str(self.base_path)
assert result == str(self.base_path), "Should return base path when no folder is provided"
def test_calc_download_path_with_folder(self):
"""Test calculating download path with folder."""
folder = "test_folder"
result = calc_download_path(str(self.base_path), folder, create_path=False)
expected = str(self.base_path / folder)
assert result == expected
assert result == expected, "Should append folder to base path"
def test_calc_download_path_creates_directory(self):
"""Test that the function creates directories when create_path=True."""
folder = "new_folder"
result = calc_download_path(str(self.base_path), folder, create_path=True)
expected_path = self.base_path / folder
assert result == str(expected_path)
assert expected_path.exists()
assert result == str(expected_path), "Should return the new path"
assert expected_path.exists(), "Directory should be created"
def test_calc_download_path_path_object(self):
"""Test with Path object as input."""
folder = "test_folder"
result = calc_download_path(self.base_path, folder, create_path=False)
expected = str(self.base_path / folder)
assert result == expected
assert result == expected, "Should handle Path object for base path"
def test_calc_download_path_nested_folder(self):
"""Test with nested folder structure."""
folder = "parent/child"
result = calc_download_path(str(self.base_path), folder, create_path=True)
expected_path = self.base_path / "parent" / "child"
assert result == str(expected_path)
assert expected_path.exists()
assert result == str(expected_path), "Should handle nested folder structure"
assert expected_path.exists(), "Nested directories should be created"
def test_calc_download_path_none_folder(self):
"""Test with None folder."""
result = calc_download_path(str(self.base_path), None, create_path=False)
assert result == str(self.base_path)
assert result == str(self.base_path), "Should return base path when folder is None"
def test_calc_download_path_removes_leading_slash(self):
"""Test that leading slash is removed from folder."""
folder = "/test_folder"
result = calc_download_path(str(self.base_path), folder, create_path=False)
expected = str(self.base_path / "test_folder")
assert result == expected, "Should remove leading slash from folder"
def test_calc_download_path_path_traversal_dotdot(self):
"""Test path traversal prevention with .. sequences."""
folder = "../outside"
with pytest.raises(Exception, match="must resolve inside the base download folder"):
calc_download_path(str(self.base_path), folder, create_path=False)
def test_calc_download_path_path_traversal_nested_dotdot(self):
"""Test path traversal prevention with nested .. sequences."""
folder = "safe/../../outside"
with pytest.raises(Exception, match="must resolve inside the base download folder"):
calc_download_path(str(self.base_path), folder, create_path=False)
def test_calc_download_path_path_traversal_multiple_dotdot(self):
"""Test path traversal prevention with multiple .. sequences."""
folder = "../../../etc/passwd"
with pytest.raises(Exception, match="must resolve inside the base download folder"):
calc_download_path(str(self.base_path), folder, create_path=False)
def test_calc_download_path_path_traversal_absolute_path(self):
"""Test that absolute paths are made safe by removing leading slash."""
folder = "/etc/passwd"
result = calc_download_path(str(self.base_path), folder, create_path=False)
expected = str(self.base_path / "etc/passwd")
assert result == expected, "Should remove leading slash and treat as relative path"
def test_calc_download_path_path_traversal_absolute_with_dotdot(self):
"""Test path traversal with absolute path containing .. sequences."""
folder = "/../../../etc/passwd"
with pytest.raises(Exception, match="must resolve inside the base download folder"):
calc_download_path(str(self.base_path), folder, create_path=False)
def test_calc_download_path_path_traversal_mixed(self):
"""Test path traversal prevention with mixed safe/unsafe paths."""
folder = "safe/../../../unsafe"
with pytest.raises(Exception, match="must resolve inside the base download folder"):
calc_download_path(str(self.base_path), folder, create_path=False)
def test_calc_download_path_path_traversal_url_encoded(self):
"""Test path traversal prevention with URL encoded sequences."""
folder = "safe%2F..%2F..%2Funsafe" # safe/../unsafe encoded
# This should be handled at a higher level, but let's test it anyway
result = calc_download_path(str(self.base_path), folder, create_path=False)
expected = str(self.base_path / folder) # Should be treated as literal filename
assert result == expected, "URL encoded sequences should be treated as literal"
def test_calc_download_path_safe_nested_paths(self):
"""Test that legitimate nested paths work correctly."""
folder = "videos/2024/january"
result = calc_download_path(str(self.base_path), folder, create_path=True)
expected_path = self.base_path / "videos" / "2024" / "january"
assert result == str(expected_path), "Should handle legitimate nested paths"
assert expected_path.exists(), "Nested directories should be created"
def test_calc_download_path_safe_dotfiles(self):
"""Test that paths with legitimate dot files work."""
folder = ".hidden/folder"
result = calc_download_path(str(self.base_path), folder, create_path=True)
expected_path = self.base_path / ".hidden" / "folder"
assert result == str(expected_path), "Should handle dot files correctly"
assert expected_path.exists(), "Hidden directories should be created"
def test_calc_download_path_empty_folder(self):
"""Test with empty string folder."""
folder = ""
result = calc_download_path(str(self.base_path), folder, create_path=False)
assert result == str(self.base_path), "Should return base path for empty folder"
def test_calc_download_path_whitespace_folder(self):
"""Test with whitespace-only folder."""
folder = " "
result = calc_download_path(str(self.base_path), folder, create_path=True)
expected_path = self.base_path / " "
assert result == str(expected_path), "Should handle whitespace folder names"
def test_calc_download_path_unicode_folder(self):
"""Test with Unicode characters in folder name."""
folder = "测试文件夹/русский/العربية"
result = calc_download_path(str(self.base_path), folder, create_path=True)
expected_path = self.base_path / "测试文件夹" / "русский" / "العربية"
assert result == str(expected_path), "Should handle Unicode folder names"
assert expected_path.exists(), "Unicode directories should be created"
def test_calc_download_path_special_characters(self):
"""Test with special characters in folder name."""
folder = "folder-with_special.chars(123)"
result = calc_download_path(str(self.base_path), folder, create_path=True)
expected_path = self.base_path / folder
assert result == str(expected_path), "Should handle special characters"
assert expected_path.exists(), "Directory with special chars should be created"
def test_calc_download_path_null_byte_attack(self):
"""Test that null bytes are prevented."""
folder = "folder\x00../../../etc/passwd"
# Any exception is acceptable for null byte attacks
with pytest.raises((ValueError, Exception)):
calc_download_path(str(self.base_path), folder, create_path=False)
def test_calc_download_path_null_byte_at_end(self):
"""Test that null bytes at end are prevented."""
folder = "../../../etc/passwd\x00"
# Any exception is acceptable for null byte attacks
with pytest.raises((ValueError, Exception)):
calc_download_path(str(self.base_path), folder, create_path=False)
def test_calc_download_path_newline_attack(self):
"""Test that newlines in path traversal are prevented."""
folder = "folder\n../../../etc/passwd"
with pytest.raises(Exception, match="must resolve inside the base download folder"):
calc_download_path(str(self.base_path), folder, create_path=False)
def test_calc_download_path_carriage_return_attack(self):
"""Test that carriage returns in path traversal are prevented."""
folder = "folder\r../../../etc/passwd"
with pytest.raises(Exception, match="must resolve inside the base download folder"):
calc_download_path(str(self.base_path), folder, create_path=False)
def test_calc_download_path_tab_attack(self):
"""Test that tabs in path traversal are prevented."""
folder = "folder\t../../../etc/passwd"
with pytest.raises(Exception, match="must resolve inside the base download folder"):
calc_download_path(str(self.base_path), folder, create_path=False)
def test_calc_download_path_vertical_tab_attack(self):
"""Test that vertical tabs in path traversal are prevented."""
folder = "folder\x0b../../../etc/passwd"
with pytest.raises(Exception, match="must resolve inside the base download folder"):
calc_download_path(str(self.base_path), folder, create_path=False)
def test_calc_download_path_form_feed_attack(self):
"""Test that form feeds in path traversal are prevented."""
folder = "folder\x0c../../../etc/passwd"
with pytest.raises(Exception, match="must resolve inside the base download folder"):
calc_download_path(str(self.base_path), folder, create_path=False)
def test_calc_download_path_url_encoded_safe(self):
"""Test that URL encoded sequences are treated as literals (safe)."""
folder = "folder%00../../../etc/passwd" # %00 is URL encoded null
with pytest.raises(Exception, match="must resolve inside the base download folder"):
calc_download_path(str(self.base_path), folder, create_path=False)
def test_calc_download_path_url_encoded_traversal_safe(self):
"""Test that URL encoded path traversal is treated as literal (safe)."""
folder = "folder..%2F..%2F..%2Fetc%2Fpasswd" # ..%2F = ../ encoded
result = calc_download_path(str(self.base_path), folder, create_path=False)
expected = str(self.base_path / folder) # Should be treated as literal filename
assert result == expected, "URL encoded sequences should be treated as literal filename"
def test_calc_download_path_backslash_attack(self):
"""Test that backslashes in path traversal are handled correctly."""
folder = "folder\\..\\..\\..\\etc\\passwd"
# On Unix systems, backslashes are treated as literal characters in filenames
result = calc_download_path(str(self.base_path), folder, create_path=False)
expected = str(self.base_path / folder)
assert result == expected, "Backslashes should be treated as literal characters on Unix"
def test_calc_download_path_mixed_separators_attack(self):
"""Test path traversal with mixed separators."""
folder = "folder/../../../etc/passwd"
with pytest.raises(Exception, match="must resolve inside the base download folder"):
calc_download_path(str(self.base_path), folder, create_path=False)
class TestMergeDict:
@ -174,6 +345,303 @@ class TestMergeDict:
result = merge_dict({}, {})
assert result == {}
# Parameter Pollution Security Tests
def test_merge_dict_blocks_class_pollution(self):
"""Test that __class__ attribute pollution is blocked."""
source = {"__class__": "malicious_class", "safe": "value"}
destination = {"existing": "data"}
result = merge_dict(source, destination)
assert "__class__" not in result, "__class__ should be filtered out"
assert result["safe"] == "value", "Safe values should be preserved"
assert result["existing"] == "data", "Existing data should be preserved"
def test_merge_dict_blocks_dict_pollution(self):
"""Test that __dict__ attribute pollution is blocked."""
source = {"__dict__": {"injected": "payload"}, "normal": "data"}
destination = {"target": "value"}
result = merge_dict(source, destination)
assert "__dict__" not in result, "__dict__ should be filtered out"
assert result["normal"] == "data", "Normal data should be preserved"
assert result["target"] == "value", "Target data should be preserved"
def test_merge_dict_blocks_globals_pollution(self):
"""Test that __globals__ attribute pollution is blocked."""
source = {"__globals__": {"malicious": "code"}, "data": "safe"}
destination = {"existing": "value"}
result = merge_dict(source, destination)
assert "__globals__" not in result, "__globals__ should be filtered out"
assert result["data"] == "safe", "Safe data should be preserved"
def test_merge_dict_blocks_builtins_pollution(self):
"""Test that __builtins__ attribute pollution is blocked."""
source = {"__builtins__": {"eval": "dangerous"}, "normal": "value"}
destination = {"target": "data"}
result = merge_dict(source, destination)
assert "__builtins__" not in result, "__builtins__ should be filtered out"
assert result["normal"] == "value", "Normal values should be preserved"
def test_merge_dict_blocks_multiple_dunder_pollution(self):
"""Test that multiple dangerous dunder attributes are blocked."""
source = {
"__class__": "malicious",
"__dict__": {"bad": "data"},
"__globals__": {"evil": "code"},
"__builtins__": {"dangerous": "function"},
"safe_key": "safe_value",
}
destination = {"existing": "data"}
result = merge_dict(source, destination)
# All dangerous attributes should be filtered out
dangerous_keys = ["__class__", "__dict__", "__globals__", "__builtins__"]
for key in dangerous_keys:
assert key not in result, f"{key} should be filtered out"
# Safe data should be preserved
assert result["safe_key"] == "safe_value"
assert result["existing"] == "data"
def test_merge_dict_nested_dunder_pollution(self):
"""Test that nested dangerous attributes are handled correctly."""
source = {"nested": {"__class__": "malicious_nested", "safe_nested": "value"}, "normal": "data"}
destination = {"nested": {"existing_nested": "original"}}
result = merge_dict(source, destination)
# Nested dangerous attributes should be filtered out
assert "__class__" not in result["nested"], "Nested __class__ should be filtered"
# Safe nested data should be preserved
assert result["nested"]["safe_nested"] == "value"
assert result["nested"]["existing_nested"] == "original"
def test_merge_dict_prototype_pollution_attempt(self):
"""Test protection against prototype pollution attempts."""
source = {"__proto__": {"polluted": True}, "constructor": {"prototype": {"polluted": True}}, "safe": "data"}
destination = {"existing": "value"}
result = merge_dict(source, destination)
# These should be treated as regular keys (not filtered unless explicitly in the filter list)
# The function filters Python-specific dangerous attributes, not JavaScript ones
assert result["safe"] == "data"
assert result["existing"] == "value"
def test_merge_dict_special_method_pollution(self):
"""Test with various Python special methods."""
source = {
"__init__": "malicious_init",
"__new__": "malicious_new",
"__call__": "malicious_call",
"__getattr__": "malicious_getattr",
"__setattr__": "malicious_setattr",
"safe": "value",
}
destination = {"target": "data"}
result = merge_dict(source, destination)
# These are not in the current filter list, so they should pass through
# This test documents current behavior - may need updating if more filters are added
assert result["safe"] == "value"
assert result["target"] == "data"
def test_merge_dict_list_pollution_safe(self):
"""Test that list merging doesn't allow dangerous manipulation."""
source = {"items": ["new1", "new2"]}
destination = {"items": ["old1", "old2"]}
result = merge_dict(source, destination)
# Lists should be concatenated safely (destination + source)
assert result["items"] == ["old1", "old2", "new1", "new2"]
def test_merge_dict_deep_nested_pollution(self):
"""Test with deeply nested dangerous attributes."""
source = {
"level1": {
"level2": {
"__class__": "deep_malicious",
"level3": {"__globals__": "very_deep_malicious"},
"safe_deep": "value",
}
}
}
destination = {"level1": {"level2": {"existing": "data"}}}
result = merge_dict(source, destination)
# The function should now properly filter all dangerous keys recursively
assert "__class__" not in result["level1"]["level2"], "Deep __class__ should be filtered"
assert "__globals__" not in result["level1"]["level2"]["level3"], "Very deep __globals__ should be filtered"
# Safe data should be preserved
assert result["level1"]["level2"]["safe_deep"] == "value"
assert result["level1"]["level2"]["existing"] == "data"
def test_merge_dict_type_validation(self):
"""Test that non-dict parameters are properly rejected."""
# Test with non-dict source
with pytest.raises(TypeError, match="Both source and destination must be dictionaries"):
merge_dict("not_a_dict", {"key": "value"})
# Test with non-dict destination
with pytest.raises(TypeError, match="Both source and destination must be dictionaries"):
merge_dict({"key": "value"}, "not_a_dict")
# Test with both non-dict
with pytest.raises(TypeError, match="Both source and destination must be dictionaries"):
merge_dict("not_a_dict", ["also_not_dict"])
def test_merge_dict_immutability(self):
"""Test that original dictionaries are not modified (immutability)."""
original_source = {"a": 1, "nested": {"b": 2}}
original_destination = {"c": 3, "nested": {"d": 4}}
# Make copies to compare later
source_copy = copy.deepcopy(original_source)
destination_copy = copy.deepcopy(original_destination)
result = merge_dict(original_source, original_destination)
# Original dictionaries should be unchanged
assert original_source == source_copy, "Source should not be modified"
assert original_destination == destination_copy, "Destination should not be modified"
# Result should be different from both originals
assert result != original_source
assert result != original_destination
def test_merge_dict_custom_max_depth(self):
"""Test custom max_depth parameter."""
# Create a deep nested structure
deep_source = {}
current = deep_source
for _ in range(10):
current["level"] = {}
current = current["level"]
current["data"] = "deep_value"
# Test with default max_depth (50) - should work
result = merge_dict(deep_source, {})
assert self._get_nested_value(result, ["level"] * 10 + ["data"]) == "deep_value"
# Test with custom max_depth (5) - should raise RecursionError
with pytest.raises(RecursionError, match="Recursion depth limit exceeded \\(5\\)"):
merge_dict(deep_source, {}, max_depth=5)
# Test with higher max_depth (20) - should work
result = merge_dict(deep_source, {}, max_depth=20)
assert self._get_nested_value(result, ["level"] * 10 + ["data"]) == "deep_value"
def test_merge_dict_custom_max_list_size(self):
"""Test custom max_list_size parameter."""
large_list = list(range(5000))
source = {"data": large_list}
# Test with default max_list_size (10000) - should preserve full list
result = merge_dict(source, {})
assert len(result["data"]) == 5000
assert result["data"] == large_list
# Test with custom max_list_size (1000) - should truncate
result = merge_dict(source, {}, max_list_size=1000)
assert len(result["data"]) == 1000
assert result["data"] == large_list[:1000]
# Test with higher max_list_size (10000) - should preserve full list
result = merge_dict(source, {}, max_list_size=10000)
assert len(result["data"]) == 5000
assert result["data"] == large_list
def test_merge_dict_list_merging_with_size_limits(self):
"""Test list merging respects size limits."""
source = {"items": list(range(3000))}
destination = {"items": list(range(2000, 5000))} # 3000 items
# Total would be 6000 items, but limit is 4000
result = merge_dict(source, destination, max_list_size=4000)
# Should have original destination (3000) + truncated source (1000) = 4000
assert len(result["items"]) == 4000
# First 3000 should be from destination
assert result["items"][:3000] == list(range(2000, 5000))
# Next 1000 should be from source (truncated)
assert result["items"][3000:] == list(range(1000))
def test_merge_dict_nested_with_limits(self):
"""Test nested merging with both depth and size limits."""
# Create nested structure that exceeds depth limit
deep_source = {"level1": {"level2": {"level3": {"level4": {"data": "deep"}}}}}
# Create structure with large list
list_source = {"level1": {"large_data": list(range(2000))}}
destination = {"level1": {"existing": "data"}}
# Test with restrictive limits
result = merge_dict(list_source, destination, max_depth=10, max_list_size=1000)
assert result["level1"]["existing"] == "data"
assert len(result["level1"]["large_data"]) == 1000
assert result["level1"]["large_data"] == list(range(1000))
# Test depth limit exceeded
with pytest.raises(RecursionError):
merge_dict(deep_source, destination, max_depth=3)
def test_merge_dict_zero_limits(self):
"""Test behavior with zero limits."""
source = {"data": [1, 2, 3]}
destination = {"existing": "value"}
# Zero max_list_size should result in empty lists
result = merge_dict(source, destination, max_list_size=0)
assert result["existing"] == "value"
assert result["data"] == [] # Truncated to empty
# Zero max_depth should fail immediately on any nesting
with pytest.raises(RecursionError):
merge_dict({"nested": {"data": "value"}}, {}, max_depth=0)
def test_merge_dict_extreme_limits(self):
"""Test with very high limits."""
# Create moderately nested structure
source = {"a": {"b": {"c": {"data": "nested"}}}}
# Very high limits should work normally
result = merge_dict(source, {}, max_depth=10000, max_list_size=1000000)
assert result["a"]["b"]["c"]["data"] == "nested"
def test_merge_dict_limits_with_circular_reference_protection(self):
"""Test that limits work together with circular reference protection."""
source = {"data": {}}
source["data"]["circular"] = source # Create circular reference
# Should fail with ValueError (circular reference) before hitting depth limit
with pytest.raises(ValueError, match="Circular reference detected"):
merge_dict(source, {}, max_depth=100)
def test_merge_dict_backward_compatibility(self):
"""Test that existing code works without specifying new parameters."""
source = {"a": 1, "nested": {"b": 2}}
destination = {"c": 3, "nested": {"d": 4}}
# Should work exactly as before with default parameters
result = merge_dict(source, destination)
assert result["a"] == 1
assert result["c"] == 3
assert result["nested"]["b"] == 2
assert result["nested"]["d"] == 4
def _get_nested_value(self, data: dict, keys: list):
"""Helper to get value from deeply nested dict."""
current = data
for key in keys:
current = current[key]
return current
class TestIsPrivateAddress:
"""Test the is_private_address function."""
@ -590,6 +1058,7 @@ class TestCheckId:
def teardown_method(self):
"""Clean up after tests."""
import shutil
shutil.rmtree(self.temp_dir, ignore_errors=True)
def test_check_id_with_youtube_id(self):
@ -620,8 +1089,11 @@ class TestArgConverter:
def test_arg_converter_basic(self):
"""Test basic arg_converter functionality."""
try:
result = arg_converter("--quiet")
result = arg_converter("--quiet --match-filters 'duration<2min' --download-archive archive.txt")
assert isinstance(result, dict)
assert result.get("quiet") is True, "quiet should be True"
assert result.get("download_archive") == "archive.txt"
assert "match_filter" in result, "match_filters should be in result"
except (ModuleNotFoundError, AttributeError, ImportError):
# Expected when yt_dlp is not available or differs
assert True
@ -650,6 +1122,7 @@ class TestGetPossibleImages:
def teardown_method(self):
"""Clean up after tests."""
import shutil
shutil.rmtree(self.temp_dir, ignore_errors=True)
def test_get_possible_images(self):
@ -706,6 +1179,7 @@ class TestGetFile:
def teardown_method(self):
"""Clean up after tests."""
import shutil
shutil.rmtree(self.temp_dir, ignore_errors=True)
def test_get_file_exists(self):
@ -775,6 +1249,7 @@ class TestGetFiles:
def teardown_method(self):
"""Clean up after tests."""
import shutil
shutil.rmtree(self.temp_dir, ignore_errors=True)
def test_get_files_root(self):
@ -800,10 +1275,12 @@ class TestReadLogfile:
def teardown_method(self):
"""Clean up after tests."""
import shutil
shutil.rmtree(self.temp_dir, ignore_errors=True)
def test_read_logfile_nonexistent(self):
"""Test reading non-existent log file."""
async def test():
result = await read_logfile(self.log_file)
assert isinstance(result, dict)
@ -834,6 +1311,7 @@ class TestTailLog:
def teardown_method(self):
"""Clean up after tests."""
import shutil
shutil.rmtree(self.temp_dir, ignore_errors=True)
def test_tail_log_nonexistent(self):
@ -864,6 +1342,7 @@ class TestLoadCookies:
def teardown_method(self):
"""Clean up after tests."""
import shutil
shutil.rmtree(self.temp_dir, ignore_errors=True)
def test_load_cookies_invalid_file(self):
@ -974,6 +1453,7 @@ class TestInitClass:
def test_init_class_basic(self):
"""Test basic class initialization."""
@dataclass
class TestClass:
name: str = ""
@ -1006,6 +1486,7 @@ class TestLoadModules:
def teardown_method(self):
"""Clean up after tests."""
import shutil
shutil.rmtree(self.temp_dir, ignore_errors=True)
def test_load_modules_basic(self):

6
pyproject.toml
View file

@ -157,6 +157,7 @@ ignore = [
"LOG015",
"PLC0415",
"S603",
"D203",
]
# Allow fix for all enabled rules (when `--fix`) is provided.
@ -203,3 +204,8 @@ link-mode = "copy"
pythonpath = ["."]
testpaths = ["app/tests"]
addopts = "-v --tb=short"
[dependency-groups]
dev = [
"pytest>=8.4.2",
]

7
ui/app/components/ConditionForm.vue
View file

@ -505,7 +505,8 @@ const addExtra = (): void => {
}
const removeExtra = (key: string): void => {
delete form.extras[key]
const { [key]: _, ...rest } = form.extras
form.extras = rest
}
const updateExtraKey = (event: Event, oldKey: string): void => {
@ -530,8 +531,8 @@ const updateExtraKey = (event: Event, oldKey: string): void => {
}
const value = form.extras[oldKey]
delete form.extras[oldKey]
form.extras[newKey] = value
const { [oldKey]: _, ...rest } = form.extras
form.extras = { ...rest, [newKey]: value }
}
}

2
ui/app/components/FloatingImage.vue
View file

@ -35,7 +35,7 @@ const url = ref<string | null>(null)
const error = ref(false)
const isPreloading = ref(false)
let loadTimer: ReturnType<typeof setTimeout> | null = null
const loadTimer: ReturnType<typeof setTimeout> | null = null
const cancelRequest = new AbortController()
const defaultLoader = async (): Promise<void> => {

12
ui/app/components/History.vue
View file

@ -615,7 +615,7 @@ const deleteSelectedItems = async () => {
}
const clearCompleted = async () => {
let msg = 'Clear all completed downloads?'
const msg = 'Clear all completed downloads?'
if (false === (await box.confirm(msg))) {
return
}
@ -774,7 +774,7 @@ const removeItem = async (item: StoreItem) => {
}
const retryItem = (item: StoreItem, re_add = false) => {
let item_req: Partial<StoreItem> = {
const item_req: Partial<StoreItem> = {
url: item.url,
preset: item.preset,
folder: item.folder,
@ -801,10 +801,12 @@ const retryItem = (item: StoreItem, re_add = false) => {
const pImg = (e: Event) => {
const target = e.target as HTMLImageElement
target.naturalHeight > target.naturalWidth && target.classList.add('image-portrait')
if (target.naturalHeight > target.naturalWidth) {
target.classList.add('image-portrait')
}
}
watch(video_item, v => {
watch(video_item, (v) => {
if (!bg_enable.value) {
return
}
@ -823,7 +825,7 @@ const downloadSelected = async () => {
toast.error('No items selected.')
return
}
let files_list: string[] = []
const files_list: string[] = []
for (const key in selectedElms.value) {
const item_id = selectedElms.value[key]
if (!item_id) {

5
ui/app/components/LateLoader.vue
View file

@ -29,8 +29,9 @@ function onIdle(cb: () => void): void {
}
}
const { stop } = useIntersectionObserver(targetEl, ([{ isIntersecting }]) => {
if (isIntersecting) {
const { stop } = useIntersectionObserver(targetEl, (entries) => {
const entry = entries[0]
if (entry?.isIntersecting) {
if (unrenderTimer) clearTimeout(unrenderTimer)
renderTimer = setTimeout(() => { shouldRender.value = true }, props.unrender ? 200 : 0)

2
ui/app/components/NotificationForm.vue
View file

@ -377,7 +377,7 @@ const checkInfo = async () => {
if (!isApprise.value) {
try {
new URL(form.request.url)
} catch (_) {
} catch {
toast.error('Invalid URL')
return
}

10
ui/app/components/Queue.vue
View file

@ -427,7 +427,6 @@ const setStatus = (item: StoreItem): string => {
return 'Streaming'
}
if ('preparing' === item.status) {
// @ts-ignore
return ag(item, 'extras.external_downloader') ? 'External-DL' : 'Preparing..'
}
if (!item.status) {
@ -494,7 +493,6 @@ const updateProgress = (item: StoreItem): string => {
return 'Post-processors are running.'
}
if ('preparing' === item.status) {
// @ts-ignore
return ag(item, 'extras.external_downloader') ? 'External downloader.' : 'Preparing'
}
if (null != item.status) {
@ -547,7 +545,7 @@ const startItems = async () => {
if (1 > selectedElms.value.length) {
return
}
let filtered: string[] = []
const filtered: string[] = []
selectedElms.value.forEach(id => {
const item = stateStore.get('queue', id) as StoreItem
if (item && !item.auto_start && !item.status) {
@ -569,7 +567,7 @@ const pauseSelected = async () => {
if (1 > selectedElms.value.length) {
return
}
let filtered: string[] = []
const filtered: string[] = []
selectedElms.value.forEach(id => {
const item = stateStore.get('queue', id) as StoreItem
if (item && item.auto_start && !item.status) {
@ -589,7 +587,9 @@ const pauseSelected = async () => {
const pImg = (e: Event) => {
const target = e.target as HTMLImageElement
target.naturalHeight > target.naturalWidth ? target.classList.add('image-portrait') : null
if (target.naturalHeight > target.naturalWidth) {
target.classList.add('image-portrait')
}
}
watch(embed_url, v => {

4
ui/app/components/VideoPlayer.vue
View file

@ -238,14 +238,14 @@ onMounted(async () => {
sources.value.push({
src,
type: allowedCodec ? response.mimetype : 'application/x-mpegURL',
onerror: (e: Event) => src_error(),
onerror: (_e: Event) => src_error(),
})
} else {
const src = makeDownload(config, props.item, 'api/download')
sources.value.push({
src,
type: response.mimetype,
onerror: e => src_error(),
onerror: (_e: Event) => src_error(),
})
}

2
ui/app/composables/useNotification.ts
View file

@ -16,6 +16,7 @@ export interface notificationOptions {
force?: boolean,
closeOnClick?: boolean,
position?: POSITION
// eslint-disable-next-line @typescript-eslint/no-unsafe-function-type
onClick?: (closeToast: Function) => void
store?: boolean,
lowPriority?: boolean
@ -52,6 +53,7 @@ function notify(type: notificationType, message: string, opts?: notificationOpti
opts.closeOnClick = toastDismissOnClick.value
opts.position = toastPosition.value ?? POSITION.TOP_RIGHT
// eslint-disable-next-line @typescript-eslint/no-unsafe-function-type
opts.onClick = (closeToast: Function) => {
if (opts?.closeOnClick !== false) {
closeToast()

10
ui/app/layouts/default.vue
View file

@ -251,7 +251,7 @@ const applyPreferredColorScheme = (scheme: string) => {
onMounted(async () => {
try {
await handleImage(bg_enable.value)
} catch (e) { }
} catch { }
try {
const opts = await request('/api/yt-dlp/options')
@ -260,11 +260,11 @@ onMounted(async () => {
}
const data: Array<YTDLPOption> = await opts.json()
config.ytdlp_options = data
} catch (e) { }
} catch { }
try {
applyPreferredColorScheme(selectedTheme.value)
} catch (e) { }
} catch { }
})
watch(selectedTheme, value => {
@ -274,7 +274,7 @@ watch(selectedTheme, value => {
return
}
applyPreferredColorScheme(value)
} catch (e) { }
} catch { }
})
const reloadPage = () => window.location.reload()
@ -362,7 +362,7 @@ const loadImage = async (force = false) => {
}
}
const changeRoute = async (_: MouseEvent, callback: Function | null = null) => {
const changeRoute = async (_: MouseEvent, callback: (() => void) | null = null) => {
showMenu.value = false
document.querySelectorAll('div.has-dropdown').forEach(el => el.classList.remove('is-active'))
if (callback) {

22
ui/app/pages/browser/[...slug].vue
View file

@ -262,7 +262,7 @@ const show_filter = ref<boolean>(false)
const reset_dialog = () => ({
visible: false,
title: 'Confirm Action',
confirm: (opts: any) => { },
confirm: (_opts: any) => { },
message: '',
html_message: '',
options: [],
@ -317,8 +317,8 @@ const sortedItems = (items: FileItem[]): FileItem[] => {
if (sort_by.value === 'date') {
return items.sort((a, b) => {
let aDate = new Date(a.mtime)
let bDate = new Date(b.mtime)
const aDate = new Date(a.mtime)
const bDate = new Date(b.mtime)
return 'asc' === sort_order.value ? aDate.getTime() - bDate.getTime() : bDate.getTime() - aDate.getTime()
})
}
@ -479,7 +479,7 @@ const makeBreadCrumb = (path: string): { name: string, link: string, path: strin
path = path.replace(/^\/+/, '').replace(/\/+$/, '')
let links = []
const links = []
links.push({
name: 'Home',
link: baseLink,
@ -487,9 +487,9 @@ const makeBreadCrumb = (path: string): { name: string, link: string, path: strin
})
// -- explode path and create links
let parts = path.split('/')
const parts = path.split('/')
parts.forEach((part, index) => {
let path = baseLink + parts.slice(0, index + 1).join('/')
const path = baseLink + parts.slice(0, index + 1).join('/')
links.push({
name: part,
link: path,
@ -570,12 +570,12 @@ const createDirectory = async (dir: string): Promise<void> => {
return
}
let new_dir = sTrim(newDir, '/')
const new_dir = sTrim(newDir, '/')
if (!new_dir || new_dir === dir) {
return
}
await actionRequest({ path: dir || '/' } as FileItem, 'directory', { new_dir: new_dir }, (item, action, data) => {
await actionRequest({ path: dir || '/' } as FileItem, 'directory', { new_dir: new_dir }, (_item, _action, _data) => {
reloadContent(path.value, true)
toast.success(`Successfully created '${new_dir}'.`)
})
@ -598,7 +598,7 @@ const handleAction = async (action: string, item: FileItem): Promise<void> => {
return
}
let new_name = newName.trim()
const new_name = newName.trim()
if (!new_name || new_name === item.name) {
return
}
@ -625,7 +625,7 @@ const handleAction = async (action: string, item: FileItem): Promise<void> => {
return
}
await actionRequest(item, 'delete', {}, (item, action, data) => {
await actionRequest(item, 'delete', {}, (item, _action, _data) => {
items.value = items.value.filter(i => i.path !== item.path)
toast.warning(`Deleted '${item.name}'.`)
})
@ -646,7 +646,7 @@ const handleAction = async (action: string, item: FileItem): Promise<void> => {
return
}
let new_path = sTrim(newPath, '/') || '/'
const new_path = sTrim(newPath, '/') || '/'
if (!new_path || new_path === item.path) {
return
}

5
ui/app/pages/notifications.vue
View file

@ -403,10 +403,11 @@ const exportItem = async (item: notification) => {
_version: '1.0',
}
const keys = ['id', 'raw']
const keys = ['id', 'raw'] as const
keys.forEach(k => {
if (Object.prototype.hasOwnProperty.call(data, k)) {
delete (data as any)[k]
const { [k]: _, ...rest } = data as any
Object.assign(data, rest)
}
})

3
ui/app/pages/presets.vue
View file

@ -379,7 +379,8 @@ const exportItem = (item: Preset) => {
for (const key of keys) {
if (key in data) {
delete data[key]
const { [key]: _, ...rest } = data
Object.assign(data, rest)
}
}

10
ui/app/pages/tasks.vue
View file

@ -394,7 +394,7 @@ const display_style = useStorage<string>("tasks_display_style", "cards")
const isMobile = useMediaQuery({ maxWidth: 1024 })
const tasks = ref<Array<task_item>>([])
const task = ref<task_item | Object>({})
const task = ref<task_item | Record<string, unknown>>({})
const taskRef = ref<string>('')
const toggleForm = ref<boolean>(false)
const isLoading = ref<boolean>(true)
@ -409,7 +409,7 @@ const table_container = ref(false)
const reset_dialog = () => ({
visible: false,
title: 'Confirm Action',
confirm: (opts: any) => { },
confirm: (_opts: any) => { },
message: '',
html_message: '',
options: [],
@ -649,7 +649,7 @@ onMounted(async () => {
const tryParse = (expression: string) => {
try {
return moment(CronExpressionParser.parse(expression).next().toISOString()).fromNow()
} catch (e) {
} catch {
return "Invalid"
}
}
@ -706,7 +706,7 @@ const runNow = async (item: task_item, mass: boolean = false) => {
item.in_progress = true
}
let data = {
const data = {
url: item.url,
preset: item.preset,
} as task_item
@ -754,7 +754,7 @@ const statusHandler = async (stream: string) => {
const exportItem = async (item: task_item) => {
const info = JSON.parse(JSON.stringify(item))
let data = {
const data = {
name: info.name,
url: info.url,
preset: info.preset,

6
ui/app/stores/SocketStore.ts
View file

@ -1,6 +1,4 @@
import { io } from "socket.io-client";
import type { Socket as IOSocket, SocketOptions } from "socket.io-client"
import type { ManagerOptions } from "socket.io-client";
import { io, type Socket as IOSocket, type SocketOptions, type ManagerOptions } from "socket.io-client"
import type { ConfigState } from "~/types/config";
import type { StoreItem } from "~/types/store";
@ -29,7 +27,7 @@ export const useSocketStore = defineStore('socket', () => {
}
const connect = () => {
let opts = {
const opts = {
transports: ['websocket', 'polling'],
withCredentials: true,
} as Partial<ManagerOptions & SocketOptions>

6
ui/app/stores/StateStore.ts
View file

@ -22,7 +22,8 @@ export const useStateStore = defineStore('state', () => {
const remove = (type: StateType, key: KeyType): void => {
if (state[type][key]) {
delete state[type][key]
const { [key]: _, ...rest } = state[type]
state[type] = rest
}
}
@ -45,7 +46,8 @@ export const useStateStore = defineStore('state', () => {
const move = (fromType: StateType, toType: StateType, key: KeyType): void => {
if (state[fromType][key]) {
state[toType][key] = state[fromType][key]
delete state[fromType][key]
const { [key]: _, ...rest } = state[fromType]
state[fromType] = rest
}
}

2
ui/app/types/dl_fields.d.ts vendored
View file

@ -20,7 +20,7 @@ type DLField = {
icon?: string;
/** The order of the field, used to sort the fields in the UI. */
order: number = 1;
order: number;
/** The default value of the field, It's currently unused */
value: string;

12
ui/app/types/sockets.d.ts vendored
View file

@ -1,8 +1,8 @@
export type Event = {
id: str
created_at: str
event: str
title: str | null = null
message: str | null = null
data: Any = { }
id: string
created_at: string
event: string
title: string | null
message: string | null
data: Record<string, unknown>
}

2
ui/app/utils/embedable.ts
View file

@ -8,7 +8,7 @@ const sources: EmbedSource[] = [
{
name: 'youtube',
url: 'https://www.youtube-nocookie.com/embed/{id}',
regex: /(?:https?:\/\/)?(?:www\.)?(?:youtube\.com\/(?:shorts\/|[^\/\n\s]+\/\S+\/|(?:v|e(?:mbed)?)\/|\S*?[?&]v=)|youtu\.be\/)(?<id>[a-zA-Z0-9_-]{11})/,
regex: /(?:https?:\/\/)?(?:www\.)?(?:youtube\.com\/(?:shorts\/|[^/\n\s]+\/\S+\/|(?:v|e(?:mbed)?)\/|\S*?[?&]v=)|youtu\.be\/)(?<id>[a-zA-Z0-9_-]{11})/,
},
{
name: "instagram_post",

2
ui/app/utils/index.ts
View file

@ -234,6 +234,7 @@ const request = (url: string, options: RequestInit = {}): Promise<Response> => {
* @returns A string without ANSI escape sequences.
*/
const removeANSIColors = (text: string): string => {
// eslint-disable-next-line no-control-regex
return text?.replace(/[\u001b\u009b][[()#;?]*(?:[0-9]{1,4}(?:;[0-9]{0,4})*)?[0-9A-ORZcf-nqry=><]/g, '') ?? text
}
@ -601,6 +602,7 @@ const sleep = (seconds: number): Promise<void> => new Promise(resolve => setTime
*
* @returns The result of the test function.
*/
// eslint-disable-next-line @typescript-eslint/no-unsafe-function-type
const awaiter = async (test: Function, timeout_ms: number = 20 * 1000, frequency: number = 200) => {
if (typeof (test) != "function") {
throw new Error("test should be a function in awaiter(test, [timeout_ms], [frequency])")

4
ui/app/utils/ytdlp.ts
View file

@ -94,7 +94,7 @@ function _match_one(filterPart: string, dct: Record<string, any>): boolean {
};
const cmpOps = Object.keys(COMPARISON_OPERATORS)
.map(op => op.replace(/([.*+?^${}()|\[\]\\])/g, "\\$1"))
.map(op => op.replace(/([.*+?^${}()|[\]\\])/g, "\\$1"))
.join("|");
const operatorRe = new RegExp(
@ -209,7 +209,7 @@ class MatchFilterParser {
[/\s+/, null],
];
const regex = new RegExp(
tokenSpec.map(([pat, name], i) => `(?<T${i}>${pat.source})`).join("|"),
tokenSpec.map(([pat, _name], i) => `(?<T${i}>${pat.source})`).join("|"),
"g"
);
const tokens: [string, string][] = [];

7
ui/eslint.config.js
View file

@ -33,7 +33,7 @@ export default withNuxt(
'vue/first-attribute-linebreak': 'off',
'vue/attribute-hyphenation': 'off',
'vue/v-on-event-hyphenation': 'off',
'vue/block-order':'off',
'vue/block-order': 'off',
'vue/prop-name-casing': 'off',
'vue/no-v-html': 'off',
@ -44,10 +44,11 @@ export default withNuxt(
'no-undef': 'off',
'no-unused-vars': 'off',
'vue/no-unused-vars': 'off'
'vue/no-unused-vars': 'off',
"@typescript-eslint/no-explicit-any": 'off',
}
},
{
files: ['**/*.vue'],
rules: { }
rules: {}
})

5
ui/nuxt.config.ts
View file

@ -14,8 +14,7 @@ try {
}
}
}
catch (e) {
}
catch { }
export default defineNuxtConfig({
ssr: false,
@ -77,6 +76,6 @@ export default defineNuxtConfig({
telemetry: false,
compatibilityDate: "2025-08-03",
experimental: {
checkOutdatedBuildInterval: 1000 * 60 * 60,
checkOutdatedBuildInterval: 1000 * 60 * 10,
}
})

10
ui/package.json
View file

@ -9,10 +9,12 @@
"preview": "nuxt preview",
"postinstall": "nuxt prepare",
"typecheck": "nuxt typecheck",
"lint": "eslint . --max-warnings 50",
"lint:fix": "eslint . --fix --max-warnings 50",
"lint": "eslint .",
"lint:ci": "eslint . --quiet",
"lint:fix": "eslint . --fix",
"test": "vitest run",
"test_watch": "vitest"
"test:watch": "vitest",
"test:ci": "vitest run --silent --reporter=dot"
},
"web-types": "./web-types.json",
"dependencies": {
@ -47,7 +49,9 @@
},
"devDependencies": {
"@nuxt/eslint": "^1.9.0",
"@nuxt/eslint-config": "^1.9.0",
"@typescript-eslint/parser": "^8.43.0",
"typescript": "^5.9.2",
"vitest": "^3.2.4",
"vue-eslint-parser": "^10.2.0"
}

6
ui/pnpm-lock.yaml
View file

@ -60,9 +60,15 @@ importers:
'@nuxt/eslint':
specifier: ^1.9.0
version: 1.9.0(@typescript-eslint/utils@8.43.0(eslint@9.35.0(jiti@2.5.1))(typescript@5.9.2))(@vue/compiler-sfc@3.5.21)(eslint@9.35.0(jiti@2.5.1))(magicast@0.3.5)(typescript@5.9.2)(vite@7.1.5(@types/node@22.18.1)(jiti@2.5.1)(terser@5.44.0)(yaml@2.8.1))
'@nuxt/eslint-config':
specifier: ^1.9.0
version: 1.9.0(@typescript-eslint/utils@8.43.0(eslint@9.35.0(jiti@2.5.1))(typescript@5.9.2))(@vue/compiler-sfc@3.5.21)(eslint@9.35.0(jiti@2.5.1))(typescript@5.9.2)
'@typescript-eslint/parser':
specifier: ^8.43.0
version: 8.43.0(eslint@9.35.0(jiti@2.5.1))(typescript@5.9.2)
typescript:
specifier: ^5.9.2
version: 5.9.2
vitest:
specifier: ^3.2.4
version: 3.2.4(@types/node@22.18.1)(jiti@2.5.1)(terser@5.44.0)(yaml@2.8.1)

25
ui/test_fix.mjs
View file

@ -1,25 +0,0 @@
import { match_str } from './app/utils/ytdlp'
const testData = {
"age_limit": 0,
"comment_count": 6,
"channel_id": "UC-7oMv6E4Uz2tF51w5Sj49w",
"uploader_url": "https://www.youtube.com/@PlayFramePlus"
}
// The filter that was incorrectly matching before
const filter = "channel_id = 'UCfmrcEdes7yDtEISGPM1T-A' & availability = subscriber_only"
console.log('Test data:', testData)
console.log('Filter:', filter)
console.log('Result:', match_str(filter, testData))
console.log('Expected: false')
// Test individual parts
console.log('\nTesting parts individually:')
console.log('Part 1 (channel_id = \'UCfmrcEdes7yDtEISGPM1T-A\'):', match_str("channel_id = 'UCfmrcEdes7yDtEISGPM1T-A'", testData))
console.log('Part 2 (availability = subscriber_only):', match_str("availability = subscriber_only", testData))
// Test correct channel_id
console.log('\nTesting with correct channel_id:')
console.log('Correct channel_id:', match_str("channel_id = 'UC-7oMv6E4Uz2tF51w5Sj49w'", testData))

51
uv.lock
View file

@ -626,6 +626,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/76/c6/c88e154df9c4e1a2a66ccf0005a88dfb2650c1dffb6f5ce603dfbd452ce3/idna-3.10-py3-none-any.whl", hash = "sha256:946d195a0d259cbba61165e88e65941f16e9b36ea6ddb97f00452bae8b1287d3", size = 70442 },
]
[[package]]
name = "iniconfig"
version = "2.1.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/f2/97/ebf4da567aa6827c909642694d71c9fcf53e5b504f2d96afea02718862f3/iniconfig-2.1.0.tar.gz", hash = "sha256:3abbd2e30b36733fee78f9c7f7308f2d0050e88f0087fd25c2645f63c773e1c7", size = 4793 }
wheels = [
{ url = "https://files.pythonhosted.org/packages/2c/e1/e6716421ea10d38022b952c159d5161ca1193197fb744506875fbb87ea7b/iniconfig-2.1.0-py3-none-any.whl", hash = "sha256:9deba5723312380e77435581c6bf4935c94cbfab9b1ed33ef8d238ea168eb760", size = 6050 },
]
[[package]]
name = "macholib"
version = "1.16.3"
@ -765,6 +774,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/40/4b/2028861e724d3bd36227adfa20d3fd24c3fc6d52032f4a93c133be5d17ce/platformdirs-4.4.0-py3-none-any.whl", hash = "sha256:abd01743f24e5287cd7a5db3752faf1a2d65353f38ec26d98e25a6db65958c85", size = 18654 },
]
[[package]]
name = "pluggy"
version = "1.6.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/f9/e2/3e91f31a7d2b083fe6ef3fa267035b518369d9511ffab804f839851d2779/pluggy-1.6.0.tar.gz", hash = "sha256:7dcc130b76258d33b90f61b658791dede3486c3e6bfb003ee5c9bfb396dd22f3", size = 69412 }
wheels = [
{ url = "https://files.pythonhosted.org/packages/54/20/4d324d65cc6d9205fabedc306948156824eb9f0ee1633355a8f7ec5c66bf/pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746", size = 20538 },
]
[[package]]
name = "propcache"
version = "0.3.2"
@ -877,6 +895,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/18/3d/f9441a0d798bf2b1e645adc3265e55706aead1255ccdad3856dbdcffec14/pycryptodome-3.23.0-cp37-abi3-win_arm64.whl", hash = "sha256:11eeeb6917903876f134b56ba11abe95c0b0fd5e3330def218083c7d98bbcb3c", size = 1703675 },
]
[[package]]
name = "pygments"
version = "2.19.2"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/b0/77/a5b8c569bf593b0140bde72ea885a803b82086995367bf2037de0159d924/pygments-2.19.2.tar.gz", hash = "sha256:636cb2477cec7f8952536970bc533bc43743542f70392ae026374600add5b887", size = 4968631 }
wheels = [
{ url = "https://files.pythonhosted.org/packages/c7/21/705964c7812476f378728bdf590ca4b771ec72385c533964653c68e86bdc/pygments-2.19.2-py3-none-any.whl", hash = "sha256:86540386c03d588bb81d44bc3928634ff26449851e99741617ecb9037ee5ec0b", size = 1225217 },
]
[[package]]
name = "pyinstaller"
version = "6.15.0"
@ -998,6 +1025,22 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/99/09/0fc0719162e5ad723f71d41cf336f18b6b5054d70dc0fe42ace6b4d2bdc9/pysubs2-1.8.0-py3-none-any.whl", hash = "sha256:05716f5039a9ebe32cd4d7673f923cf36204f3a3e99987f823ab83610b7035a0", size = 43516 },
]
[[package]]
name = "pytest"
version = "8.4.2"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "colorama", marker = "sys_platform == 'win32'" },
{ name = "iniconfig" },
{ name = "packaging" },
{ name = "pluggy" },
{ name = "pygments" },
]
sdist = { url = "https://files.pythonhosted.org/packages/a3/5c/00a0e072241553e1a7496d638deababa67c5058571567b92a7eaa258397c/pytest-8.4.2.tar.gz", hash = "sha256:86c0d0b93306b961d58d62a4db4879f27fe25513d4b969df351abdddb3c30e01", size = 1519618 }
wheels = [
{ url = "https://files.pythonhosted.org/packages/a8/a4/20da314d277121d6534b3a980b29035dcd51e6744bd79075a6ce8fa4eb8d/pytest-8.4.2-py3-none-any.whl", hash = "sha256:872f880de3fc3a5bdc88a11b39c9710c3497a547cfa9320bc3c5e62fbf272e79", size = 365750 },
]
[[package]]
name = "python-dateutil"
version = "2.9.0.post0"
@ -1430,6 +1473,11 @@ installer = [
{ name = "pyinstaller" },
]
[package.dev-dependencies]
dev = [
{ name = "pytest" },
]
[package.metadata]
requires-dist = [
{ name = "aiocron", specifier = ">=1.8" },
@ -1466,6 +1514,9 @@ requires-dist = [
]
provides-extras = ["installer"]
[package.metadata.requires-dev]
dev = [{ name = "pytest", specifier = ">=8.4.2" }]
[[package]]
name = "zipstream-ng"
version = "1.9.0"