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

Merge pull request #489 from arabcoders/dev

Browse source 
Feat: File browser controls improvements
This commit is contained in:
Abdulmohsen 2025-11-14 20:17:14 +03:00 committed by GitHub
commit 477dc41bb6
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
9 changed files with 1952 additions and 52 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

30
app/library/DataStore.py
View file

@ -9,6 +9,7 @@ from sqlite3 import Connection
from .Download import Download
from .ItemDTO import ItemDTO
from .operations import matches_condition
from .Utils import init_class
LOG: logging.Logger = logging.getLogger("datastore")
@ -91,16 +92,31 @@ class DataStore:
def get_item(self, **kwargs) -> Download | None:
"""
Get a specific item from the datastore based on provided attributes.
Get a specific item from the datastore based on provided attributes with optional operations.
Args:
**kwargs: Arbitrary keyword arguments representing attributes of the ItemDTO.
If no attributes are provided, the method returns None.
If any attribute matches, the corresponding Download object is returned.
Each value can be either:
- A direct value (defaults to EQUAL operation): {"title": "test"}
- A tuple of (Operation, value): {"title": (Operation.CONTAIN, "test")}
If no attributes are provided, the method returns None.
If any attribute matches, the corresponding Download object is returned.
Returns:
Download | None: The requested item if found, otherwise None.
Examples:
# Direct equality check (default)
store.get_item(title="Video 1")
# Using operations
store.get_item(title=(Operation.CONTAIN, "test"))
store.get_item(id=(Operation.EQUAL, "123"), status=(Operation.NOT_EQUAL, "error"))
# Mixed usage
store.get_item(title=(Operation.CONTAIN, "test"), folder="downloads")
"""
if not kwargs:
return None
@ -110,7 +126,7 @@ class DataStore:
continue
info = self._dict[i].info.__dict__
if any((key in info and info == value) for key, value in kwargs.items()):
if any(matches_condition(key, value, info) for key, value in kwargs.items()):
return self._dict[i]
return None
@ -118,7 +134,7 @@ class DataStore:
def get_by_id(self, id: str) -> Download | None:
return self._dict.get(id, None)
def items(self) -> list[tuple[str, Download]]:
def items(self) -> OrderedDict[tuple[str, Download]]:
return self._dict.items()
def saved_items(self) -> list[tuple[str, ItemDTO]]:
@ -234,8 +250,8 @@ class DataStore:
order = "ASC" if order == "ASC" else "DESC"
total_items = self.get_total_count()
total_pages = (total_items + per_page - 1) // per_page if total_items > 0 else 1
total_items: int = self.get_total_count()
total_pages: int = (total_items + per_page - 1) // per_page if total_items > 0 else 1
# Ensure page is within valid range.
if page > total_pages and total_items > 0:

134
app/library/Utils.py
View file

@ -763,6 +763,140 @@ def get_file_sidecar(file: Path | None = None) -> dict[dict]:
return files
def rename_file(old_path: Path, new_name: str) -> tuple[Path, list[tuple[Path, Path]]]:
"""
Rename a file along with all its sidecar files.
Args:
old_path (Path): The original file path.
new_name (str): The new name for the file (not full path, just the name).
Returns:
tuple[Path, list[tuple[Path, Path]]]: Tuple of (new_path, list of (old_sidecar_path, new_sidecar_path) tuples).
Raises:
OSError: If renaming fails.
ValueError: If new_name would cause a collision with existing files.
"""
new_path: Path = old_path.parent / new_name
if new_path.exists():
msg: str = f"Destination '{new_name}' already exists"
raise ValueError(msg)
sidecar_data: dict[str, dict] = get_file_sidecar(old_path)
sidecar_files: list[Path] = []
for category in sidecar_data.values():
sidecar_files.extend([item["file"] for item in category if "file" in item and isinstance(item["file"], Path)])
renamed_sidecars: list[tuple[Path, Path]] = []
new_stem: str = new_path.stem
rename_operations: list[tuple[Path, Path]] = []
for sidecar in sidecar_files:
sidecar_suffix: str = sidecar.name[len(old_path.stem) :] # Everything after the original stem
new_sidecar_path: Path = old_path.parent / f"{new_stem}{sidecar_suffix}"
if new_sidecar_path.exists():
msg = f"Sidecar destination '{new_sidecar_path.name}' already exists"
raise ValueError(msg)
rename_operations.append((sidecar, new_sidecar_path))
try:
renamed_main: Path = old_path.rename(new_path)
for old_sidecar, new_sidecar in rename_operations:
try:
renamed_sidecar: Path = old_sidecar.rename(new_sidecar)
renamed_sidecars.append((old_sidecar, renamed_sidecar))
except OSError as e:
LOG.error(f"Failed to rename sidecar '{old_sidecar}': {e}")
try:
renamed_main.rename(old_path)
except OSError:
LOG.error(f"Failed to rollback main file rename from '{renamed_main}' to '{old_path}'")
raise
return renamed_main, renamed_sidecars
except OSError as e:
LOG.error(f"Failed to rename '{old_path}' to '{new_path}': {e}")
raise
def move_file(old_path: Path, target_dir: Path) -> tuple[Path, list[tuple[Path, Path]]]:
"""
Move a file along with all its sidecar files to a target directory.
Args:
old_path (Path): The original file path.
target_dir (Path): The target directory to move the file to.
Returns:
tuple[Path, list[tuple[Path, Path]]]: Tuple of (new_path, list of (old_sidecar_path, new_sidecar_path) tuples).
Raises:
OSError: If moving fails.
ValueError: If target directory doesn't exist or if destination would cause collision.
"""
if not target_dir.exists():
msg: str = f"Target directory '{target_dir}' does not exist"
raise ValueError(msg)
if not target_dir.is_dir():
msg = f"Target '{target_dir}' is not a directory"
raise ValueError(msg)
new_path: Path = target_dir / old_path.name
if new_path.exists():
msg = f"Destination '{new_path}' already exists"
raise ValueError(msg)
sidecar_data: dict[str, dict] = get_file_sidecar(old_path)
sidecar_files: list[Path] = []
for category in sidecar_data.values():
sidecar_files.extend([item["file"] for item in category if "file" in item and isinstance(item["file"], Path)])
renamed_sidecars: list[tuple[Path, Path]] = []
move_operations: list[tuple[Path, Path]] = []
for sidecar in sidecar_files:
new_sidecar_path: Path = target_dir / sidecar.name
if new_sidecar_path.exists():
msg = f"Sidecar destination '{new_sidecar_path}' already exists"
raise ValueError(msg)
move_operations.append((sidecar, new_sidecar_path))
try:
moved_main: Path = old_path.rename(new_path)
for old_sidecar, new_sidecar in move_operations:
try:
moved_sidecar: Path = old_sidecar.rename(new_sidecar)
renamed_sidecars.append((old_sidecar, moved_sidecar))
except OSError as e:
LOG.error(f"Failed to move sidecar '{old_sidecar}': {e}")
try:
moved_main.rename(old_path)
# Rollback any sidecars that were already moved
for rolled_back_old, rolled_back_new in renamed_sidecars:
try:
rolled_back_new.rename(rolled_back_old)
except OSError:
LOG.error(f"Failed to rollback sidecar move from '{rolled_back_new}' to '{rolled_back_old}'")
except OSError:
LOG.error(f"Failed to rollback main file move from '{moved_main}' to '{old_path}'")
raise
return moved_main, renamed_sidecars
except OSError as e:
LOG.error(f"Failed to move '{old_path}' to '{new_path}': {e}")
raise
def get_possible_images(dir: str) -> list[dict]:
images: list = []

275
app/library/operations.py Normal file
View file

@ -0,0 +1,275 @@
from enum import Enum
from typing import Any
class Operation(str, Enum):
"""Comparison operations for filtering items."""
EQUAL = "=="
"""Exact equality comparison."""
NOT_EQUAL = "!="
"""Not equal comparison."""
CONTAIN = "in"
"""Check if value is contained in the field (substring match)."""
NOT_CONTAIN = "not_in"
"""Check if value is not contained in the field."""
GREATER_THAN = ">"
"""Greater than comparison."""
LESS_THAN = "<"
"""Less than comparison."""
GREATER_EQUAL = ">="
"""Greater than or equal comparison."""
LESS_EQUAL = "<="
"""Less than or equal comparison."""
STARTS_WITH = "startswith"
"""Check if field starts with value."""
ENDS_WITH = "endswith"
"""Check if field ends with value."""
def __str__(self) -> str:
return self.value
def matches(operation: Operation | str, haystack: Any, needle: Any) -> bool:
"""
Generic comparison function that compares two values using the specified operation.
Args:
operation: The comparison operation to perform (Operation enum or string)
haystack: The first value (usually the field value from data)
needle: The second value (usually the comparison value)
Returns:
bool: True if the comparison matches, False otherwise
Examples:
>>> matches(Operation.EQUAL, "test", "test")
True
>>> matches(Operation.CONTAIN, "Python Tutorial", "Python")
True
>>> matches(Operation.GREATER_THAN, 100, 50)
True
>>> matches("==", "test", "test")
True
"""
# Parse operation if it's a string
if isinstance(operation, str):
try:
operation = Operation(operation)
except ValueError:
operation = Operation.EQUAL
try:
if Operation.EQUAL == operation:
return haystack == needle
if Operation.NOT_EQUAL == operation:
return haystack != needle
if Operation.CONTAIN == operation:
return str(needle) in str(haystack) if haystack is not None else False
if Operation.NOT_CONTAIN == operation:
return str(needle) not in str(haystack) if haystack is not None else True
if Operation.GREATER_THAN == operation:
if haystack is None or needle is None:
return False
return haystack > needle
if Operation.LESS_THAN == operation:
if haystack is None or needle is None:
return False
return haystack < needle
if Operation.GREATER_EQUAL == operation:
if haystack is None or needle is None:
return False
return haystack >= needle
if Operation.LESS_EQUAL == operation:
if haystack is None or needle is None:
return False
return haystack <= needle
if Operation.STARTS_WITH == operation:
return str(haystack).startswith(str(needle)) if haystack is not None else False
if Operation.ENDS_WITH == operation:
return str(haystack).endswith(str(needle)) if haystack is not None else False
# Unknown operation, default to equality
return haystack == needle
except (TypeError, AttributeError):
# Comparison failed (e.g., comparing incompatible types)
return False
def matches_condition(key: str, value: tuple | str | float | bool, data: dict) -> bool:
"""
Check if a field in a dictionary matches the given condition.
This is a helper function that extracts values from a dictionary and uses the generic
matches() function to perform the comparison.
Args:
key: The field name to check in the data dictionary
value: Either:
- A direct value for equality check: "test"
- A tuple of (Operation, value): (Operation.CONTAIN, "test")
- A tuple of (str, value): ("in", "test") for backward compatibility
data: Dictionary containing the data to check against
Returns:
bool: True if the condition matches, False otherwise
Examples:
>>> data = {"title": "Python Tutorial", "size": 1000}
>>> matches_condition("title", "Python Tutorial", data)
True
>>> matches_condition("title", (Operation.CONTAIN, "Python"), data)
True
>>> matches_condition("size", (Operation.GREATER_THAN, 500), data)
True
>>> matches_condition("missing", "value", data)
False
"""
if key not in data:
return False
field_value: Any = data[key]
# Parse value to extract operation and comparison value
if isinstance(value, tuple) and len(value) == 2:
operation, compare_value = value
else:
operation = Operation.EQUAL
compare_value = value
return matches(operation, field_value, compare_value)
def matches_all(data: dict, **conditions) -> bool:
"""
Check if all conditions match (AND logic).
Args:
data: Dictionary containing the data to check against
**conditions: Keyword arguments representing conditions to check
Returns:
bool: True if all conditions match, False otherwise
Examples:
>>> data = {"title": "Python Tutorial", "size": 1000, "status": "active"}
>>> matches_all(data, title=(Operation.CONTAIN, "Python"), size=(Operation.GREATER_THAN, 500))
True
>>> matches_all(data, title="Python Tutorial", status="active")
True
"""
if not conditions:
return True
return all(matches_condition(key, value, data) for key, value in conditions.items())
def matches_any(data: dict, **conditions) -> bool:
"""
Check if any condition matches (OR logic).
Args:
data: Dictionary containing the data to check against
**conditions: Keyword arguments representing conditions to check
Returns:
bool: True if any condition matches, False if none match
Examples:
>>> data = {"title": "Python Tutorial", "size": 1000}
>>> matches_any(data, title=(Operation.CONTAIN, "Java"), size=(Operation.GREATER_THAN, 500))
True
>>> matches_any(data, title="Wrong", status="Wrong")
False
"""
if not conditions:
return False
return any(matches_condition(key, value, data) for key, value in conditions.items())
def filter_items(items: list[dict], **conditions) -> list[dict]:
"""
Filter a list of dictionaries based on conditions (AND logic).
Args:
items: List of dictionaries to filter
**conditions: Keyword arguments representing conditions to check
Returns:
list[dict]: Filtered list of dictionaries that match all conditions
Examples:
>>> items = [
... {"title": "Python Tutorial", "size": 1000},
... {"title": "JavaScript Course", "size": 2000},
... {"title": "Python Advanced", "size": 1500}
... ]
>>> filter_items(items, title=(Operation.CONTAIN, "Python"))
[{"title": "Python Tutorial", "size": 1000}, {"title": "Python Advanced", "size": 1500}]
>>> filter_items(items, size=(Operation.GREATER_THAN, 1200))
[{"title": "JavaScript Course", "size": 2000}, {"title": "Python Advanced", "size": 1500}]
"""
if not conditions:
return items
return [item for item in items if matches_all(item, **conditions)]
def find_first(items: list[dict], **conditions) -> dict | None:
"""
Find the first dictionary that matches all conditions (AND logic).
Args:
items: List of dictionaries to search
**conditions: Keyword arguments representing conditions to check
Returns:
dict | None: First matching dictionary or None if no match found
Examples:
>>> items = [
... {"title": "Python Tutorial", "size": 1000},
... {"title": "JavaScript Course", "size": 2000}
... ]
>>> find_first(items, title=(Operation.CONTAIN, "Python"))
{"title": "Python Tutorial", "size": 1000}
>>> find_first(items, title="Nonexistent")
None
"""
for item in items:
if matches_all(item, **conditions):
return item
return None
def find_all(items: list[dict], **conditions) -> list[dict]:
"""
Alias for filter_items() - find all dictionaries matching conditions.
Args:
items: List of dictionaries to search
**conditions: Keyword arguments representing conditions to check
Returns:
list[dict]: List of matching dictionaries
"""
return filter_items(items, **conditions)

89
app/routes/api/browser.py
View file

@ -9,10 +9,12 @@ from aiohttp.web import Request, Response
from app.library.cache import Cache
from app.library.config import Config
from app.library.DownloadQueue import DownloadQueue
from app.library.encoder import Encoder
from app.library.Events import EventBus, Events
from app.library.ffprobe import ffprobe
from app.library.router import route
from app.library.Utils import delete_dir, get_file, get_file_sidecar, get_files, get_mime_type
from app.library.Utils import delete_dir, get_file, get_file_sidecar, get_files, get_mime_type, move_file, rename_file
LOG: logging.Logger = logging.getLogger(__name__)
@ -175,13 +177,15 @@ async def file_browser(request: Request, config: Config, encoder: Encoder) -> Re
@route("POST", "api/file/actions", "browser.file.actions")
async def path_actions(request: Request, config: Config) -> Response:
async def path_actions(request: Request, config: Config, queue: DownloadQueue, notify: EventBus) -> Response:
"""
Browser actions.
Args:
request (Request): The request object.
config (Config): The configuration object.
queue (DownloadQueue): The download queue instance.
notify (EventBus): The event bus instance.
Returns:
Response: The response object.
@ -360,16 +364,46 @@ async def path_actions(request: Request, config: Config) -> Response:
continue
try:
renamed = path.rename(new_path)
sidecar_count: int = 0
sidecar_info: str = ""
sidecar_renamed: list[tuple[Path, Path]] = []
if path.is_dir():
renamed: Path = path.rename(new_path)
else:
renamed, sidecar_renamed = rename_file(path, new_name)
sidecar_count: int = len(sidecar_renamed)
sidecar_info: str = (
f" (with {sidecar_count} sidecar file{'s' if sidecar_count != 1 else ''})"
if sidecar_count > 0
else ""
)
LOG.info(
f"Renamed '{path.relative_to(config.download_path)}' to '{renamed.relative_to(config.download_path)}'"
f"Renamed '{path.relative_to(config.download_path)}' to '{renamed.relative_to(config.download_path)}'{sidecar_info}"
)
except OSError as e:
LOG.exception(e)
record(path, ok=False, error=str(e), action=action, extra={"item": params})
continue
except ValueError as e:
record(path, ok=False, error=str(e), action=action, extra={"item": params})
continue
else:
record(path, ok=True, action=action, extra={"new_path": renamed})
extra_info: dict[str, Path] = {"new_path": renamed}
if sidecar_renamed:
extra_info["sidecar_count"] = len(sidecar_renamed)
record(path, ok=True, action=action, extra=extra_info)
for old_sidecar, new_sidecar in sidecar_renamed:
record(old_sidecar, ok=True, action=action, extra={"new_path": new_sidecar})
if item := queue.done.get_item(filename=str(path.relative_to(config.download_path))):
item.info.filename = str(renamed.relative_to(config.download_path))
if sidecar_renamed:
item.info.get_file_sidecar()
queue.done.put(item)
notify.emit(Events.ITEM_UPDATED, data=item.info)
if "delete" == action:
try:
@ -397,7 +431,7 @@ async def path_actions(request: Request, config: Config) -> Response:
continue
raw_new: str = unquote_plus(str(new_path)).strip()
target_dir = (
target_dir: Path = (
Path(config.download_path)
if not raw_new or raw_new in ("/", ".")
else Path(config.download_path).joinpath(raw_new.lstrip("/"))
@ -411,7 +445,7 @@ async def path_actions(request: Request, config: Config) -> Response:
)
continue
root_real = Path(config.download_path).resolve()
root_real: Path = Path(config.download_path).resolve()
if not target_dir.exists() or not target_dir.is_dir():
record(
path, ok=False, error="Destination path is invalid.", action=action, extra={"new_path": target_dir}
@ -433,14 +467,49 @@ async def path_actions(request: Request, config: Config) -> Response:
continue
try:
dest = target_dir.joinpath(path.name)
path.rename(dest)
sidecar_count: int = 0
sidecar_moved: list[tuple[Path, Path]] = []
sidecar_info: str = ""
if path.is_dir():
dest: Path = target_dir.joinpath(path.name)
moved: Path = path.rename(dest)
else:
moved, sidecar_moved = move_file(path, target_dir)
sidecar_count: int = len(sidecar_moved)
sidecar_info: str = (
f" (with {sidecar_count} sidecar file{'s' if sidecar_count != 1 else ''})"
if sidecar_count > 0
else ""
)
LOG.info(
f"Moved '{path.relative_to(config.download_path)}' to '{moved.relative_to(config.download_path)}'{sidecar_info}"
)
except OSError as e:
LOG.exception(e)
record(path, ok=False, error=str(e), action=action, extra={"item": params})
continue
except ValueError as e:
record(path, ok=False, error=str(e), action=action, extra={"item": params})
continue
else:
record(path, ok=True, action=action, extra={"new_path": dest})
extra_info: dict[str, Path] = {"new_path": moved}
if sidecar_moved:
extra_info["sidecar_count"] = len(sidecar_moved)
record(path, ok=True, action=action, extra=extra_info)
for old_sidecar, new_sidecar in sidecar_moved:
record(old_sidecar, ok=True, action=action, extra={"new_path": new_sidecar})
if item := queue.done.get_item(filename=str(path.relative_to(config.download_path))):
item.info.filename = str(moved.relative_to(config.download_path))
if sidecar_moved:
item.info.get_file_sidecar()
queue.done.put(item)
notify.emit(Events.ITEM_UPDATED, data=item.info)
return web.json_response(data=operations_status, status=web.HTTPOk.status_code)

838
app/tests/test_datastore.py
View file

@ -1,5 +1,6 @@
import json
import sqlite3
from collections import OrderedDict
from dataclasses import asdict
from datetime import UTC, datetime
from email.utils import formatdate
@ -8,6 +9,7 @@ import pytest
from app.library.DataStore import DataStore, StoreType
from app.library.ItemDTO import ItemDTO
from app.library.operations import Operation
class StubDownload:
@ -26,9 +28,7 @@ class StubDownload:
def make_conn() -> sqlite3.Connection:
conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
conn.execute(
"CREATE TABLE history (id TEXT PRIMARY KEY, type TEXT, url TEXT, data TEXT, created_at TEXT)"
)
conn.execute("CREATE TABLE history (id TEXT PRIMARY KEY, type TEXT, url TEXT, data TEXT, created_at TEXT)")
return conn
@ -155,3 +155,835 @@ class TestDataStore:
# Should not raise
ok = await store.test()
assert ok is True
def test_get_item_returns_none_when_no_kwargs(self) -> None:
"""Test that get_item returns None when no kwargs provided."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
result = store.get_item()
assert result is None
def test_get_item_finds_by_single_attribute(self) -> None:
"""Test that get_item correctly finds item by a single attribute."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
# Create items with different attributes
item1 = make_item(id="vid1", url="http://example.com/1", title="Video 1", folder="folder1")
item1._id = "id1" # Override auto-generated UUID
item2 = make_item(id="vid2", url="http://example.com/2", title="Video 2", folder="folder2")
item2._id = "id2" # Override auto-generated UUID
d1 = StubDownload(info=item1)
d2 = StubDownload(info=item2)
store.put(d1)
store.put(d2)
# Test finding by title
result = store.get_item(title="Video 1")
assert result is not None
assert result.info._id == "id1"
assert result.info.title == "Video 1"
# Test finding by folder
result = store.get_item(folder="folder2")
assert result is not None
assert result.info._id == "id2"
assert result.info.folder == "folder2"
# Test finding by url
result = store.get_item(url="http://example.com/1")
assert result is not None
assert result.info._id == "id1"
def test_get_item_finds_by_multiple_attributes(self) -> None:
"""Test that get_item finds item when ANY of the provided attributes match."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
item1 = make_item(id="vid1", url="http://example.com/1", title="Video 1", folder="folder1")
item1._id = "id1"
item2 = make_item(id="vid2", url="http://example.com/2", title="Video 2", folder="folder2")
item2._id = "id2"
d1 = StubDownload(info=item1)
d2 = StubDownload(info=item2)
store.put(d1)
store.put(d2)
# Test finding by multiple attributes where one matches
result = store.get_item(title="Video 1", folder="wrong_folder")
assert result is not None
assert result.info._id == "id1"
# Test finding where second attribute matches
result = store.get_item(title="Wrong Title", folder="folder2")
assert result is not None
assert result.info._id == "id2"
def test_get_item_returns_none_when_no_match(self) -> None:
"""Test that get_item returns None when no attributes match."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
item = make_item(id="vid1", url="http://example.com/1", title="Video 1", folder="folder1")
item._id = "id1"
d = StubDownload(info=item)
store.put(d)
# Test with non-matching attribute
result = store.get_item(title="Nonexistent Video")
assert result is None
# Test with non-existent attribute key
result = store.get_item(nonexistent_field="value")
assert result is None
def test_get_item_skips_items_with_no_info(self) -> None:
"""Test that get_item skips items that have no info attribute."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
# Create a valid item
item = make_item(id="vid1", url="http://example.com/1", title="Video 1")
item._id = "id1"
d = StubDownload(info=item)
store.put(d)
# Manually add an item with None info
class BrokenDownload:
def __init__(self):
self.info = None
store._dict["broken"] = BrokenDownload()
# Should still find the valid item
result = store.get_item(title="Video 1")
assert result is not None
assert result.info._id == "id1"
def test_get_item_returns_first_match(self) -> None:
"""Test that get_item returns the first matching item when multiple match."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
# Create multiple items with same title
item1 = make_item(id="vid1", url="http://example.com/1", title="Same Title", folder="folder1")
item1._id = "id1"
item2 = make_item(id="vid2", url="http://example.com/2", title="Same Title", folder="folder2")
item2._id = "id2"
d1 = StubDownload(info=item1)
d2 = StubDownload(info=item2)
store.put(d1)
store.put(d2)
# Should return first match (note: OrderedDict maintains insertion order)
result = store.get_item(title="Same Title")
assert result is not None
assert result.info._id == "id1"
def test_init(self) -> None:
"""Test DataStore initialization."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
assert store._type == StoreType.QUEUE
assert store._connection is conn
assert isinstance(store._dict, OrderedDict)
assert len(store._dict) == 0
def test_load(self) -> None:
"""Test loading items from database into memory."""
conn = make_conn()
# Insert items directly into database
item1_data = asdict(make_item(id="vid1", url="http://example.com/1", title="Video 1"))
item1_data.pop("_id", None)
item2_data = asdict(make_item(id="vid2", url="http://example.com/2", title="Video 2"))
item2_data.pop("_id", None)
created = datetime(2024, 1, 1, 12, 0, 0, tzinfo=UTC)
conn.execute(
"INSERT INTO history (id, type, url, data, created_at) VALUES (?, ?, ?, ?, ?)",
(
"id1",
str(StoreType.QUEUE),
"http://example.com/1",
json.dumps(item1_data),
created.strftime("%Y-%m-%d %H:%M:%S"),
),
)
conn.execute(
"INSERT INTO history (id, type, url, data, created_at) VALUES (?, ?, ?, ?, ?)",
(
"id2",
str(StoreType.QUEUE),
"http://example.com/2",
json.dumps(item2_data),
created.strftime("%Y-%m-%d %H:%M:%S"),
),
)
# Create store and load
store = DataStore(StoreType.QUEUE, conn)
assert len(store._dict) == 0
store.load()
assert len(store._dict) == 2
assert "id1" in store._dict
assert "id2" in store._dict
assert store._dict["id1"].info.url == "http://example.com/1"
assert store._dict["id2"].info.url == "http://example.com/2"
def test_load_with_different_store_types(self) -> None:
"""Test that load only loads items matching the store type."""
conn = make_conn()
# Insert items with different types
item1_data = asdict(make_item(id="vid1", url="http://example.com/1"))
item1_data.pop("_id", None)
item2_data = asdict(make_item(id="vid2", url="http://example.com/2"))
item2_data.pop("_id", None)
created = datetime(2024, 1, 1, 12, 0, 0, tzinfo=UTC)
conn.execute(
"INSERT INTO history (id, type, url, data, created_at) VALUES (?, ?, ?, ?, ?)",
(
"id1",
str(StoreType.QUEUE),
"http://example.com/1",
json.dumps(item1_data),
created.strftime("%Y-%m-%d %H:%M:%S"),
),
)
conn.execute(
"INSERT INTO history (id, type, url, data, created_at) VALUES (?, ?, ?, ?, ?)",
(
"id2",
str(StoreType.HISTORY),
"http://example.com/2",
json.dumps(item2_data),
created.strftime("%Y-%m-%d %H:%M:%S"),
),
)
# Load QUEUE store - should only get queue items
queue_store = DataStore(StoreType.QUEUE, conn)
queue_store.load()
assert len(queue_store._dict) == 1
assert "id1" in queue_store._dict
# Load HISTORY store - should only get history items
history_store = DataStore(StoreType.HISTORY, conn)
history_store.load()
assert len(history_store._dict) == 1
assert "id2" in history_store._dict
def test_get_by_id(self) -> None:
"""Test getting item by ID."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
item1 = make_item(id="vid1", url="http://example.com/1", title="Video 1")
item1._id = "id1"
item2 = make_item(id="vid2", url="http://example.com/2", title="Video 2")
item2._id = "id2"
d1 = StubDownload(info=item1)
d2 = StubDownload(info=item2)
store.put(d1)
store.put(d2)
# Test getting existing items
result = store.get_by_id("id1")
assert result is not None
assert result.info._id == "id1"
assert result.info.title == "Video 1"
result = store.get_by_id("id2")
assert result is not None
assert result.info._id == "id2"
# Test getting non-existent item
result = store.get_by_id("nonexistent")
assert result is None
def test_items(self) -> None:
"""Test getting all items as list of tuples."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
# Empty store
result = store.items()
assert len(list(result)) == 0
# Add items
item1 = make_item(id="vid1", url="http://example.com/1", title="Video 1")
item1._id = "id1"
item2 = make_item(id="vid2", url="http://example.com/2", title="Video 2")
item2._id = "id2"
d1 = StubDownload(info=item1)
d2 = StubDownload(info=item2)
store.put(d1)
store.put(d2)
# Test getting all items
result = list(store.items())
assert len(result) == 2
# Verify structure (list of tuples)
ids = [item[0] for item in result]
assert "id1" in ids
assert "id2" in ids
# Verify order is maintained (OrderedDict)
assert result[0][0] == "id1"
assert result[1][0] == "id2"
def test_get_total_count_with_empty_store(self) -> None:
"""Test get_total_count with empty datastore."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
count = store.get_total_count()
assert count == 0
def test_get_total_count_with_items(self) -> None:
"""Test get_total_count with items in database."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
# Add items directly to database
created = datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S")
for i in range(5):
item_data = asdict(make_item(id=f"vid{i}"))
item_data.pop("_id", None)
conn.execute(
"INSERT INTO history (id, type, url, data, created_at) VALUES (?, ?, ?, ?, ?)",
(f"id{i}", str(StoreType.QUEUE), f"http://example.com/{i}", json.dumps(item_data), created),
)
count = store.get_total_count()
assert count == 5
def test_get_total_count_respects_store_type(self) -> None:
"""Test that get_total_count only counts items of the correct type."""
conn = make_conn()
created = datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S")
# Add 3 QUEUE items
for i in range(3):
item_data = asdict(make_item(id=f"vid{i}"))
item_data.pop("_id", None)
conn.execute(
"INSERT INTO history (id, type, url, data, created_at) VALUES (?, ?, ?, ?, ?)",
(f"q{i}", str(StoreType.QUEUE), f"http://example.com/{i}", json.dumps(item_data), created),
)
# Add 2 HISTORY items
for i in range(2):
item_data = asdict(make_item(id=f"vid{i}"))
item_data.pop("_id", None)
conn.execute(
"INSERT INTO history (id, type, url, data, created_at) VALUES (?, ?, ?, ?, ?)",
(f"h{i}", str(StoreType.HISTORY), f"http://example.com/{i}", json.dumps(item_data), created),
)
queue_store = DataStore(StoreType.QUEUE, conn)
assert queue_store.get_total_count() == 3
history_store = DataStore(StoreType.HISTORY, conn)
assert history_store.get_total_count() == 2
def test_put_with_error_status_emits_event(self) -> None:
"""Test that put() emits an event when item has error status."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
item = make_item(id="vid1")
item.status = "error"
item.error = "Test error message"
d = StubDownload(info=item)
# We can't easily test event emission without mocking EventBus
# Just verify it doesn't crash
result = store.put(d)
assert result is not None
def test_put_with_no_notify_skips_event(self) -> None:
"""Test that put() with no_notify=True doesn't emit events."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
item = make_item(id="vid1")
item.status = "error"
item.error = "Test error message"
d = StubDownload(info=item)
# Should not emit event when no_notify=True
result = store.put(d, no_notify=True)
assert result is not None
assert result.info._id == item._id
def test_delete_nonexistent_item(self) -> None:
"""Test that deleting non-existent item doesn't raise error."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
# Should not raise error
store.delete("nonexistent_id")
# Verify nothing was deleted from database
row = conn.execute("SELECT * FROM history WHERE id=?", ("nonexistent_id",)).fetchone()
assert row is None
def test_has_downloads_with_empty_dict(self) -> None:
"""Test has_downloads returns False when dict is empty."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
assert store.has_downloads() is False
def test_has_downloads_with_no_eligible_downloads(self) -> None:
"""Test has_downloads returns False when no downloads are eligible."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
# Add item that's already started
item1 = make_item(id="vid1")
store.put(StubDownload(info=item1, started=True))
# Add item with auto_start=False
item2 = make_item(id="vid2")
item2.auto_start = False
store.put(StubDownload(info=item2, started=False))
assert store.has_downloads() is False
def test_get_next_download_returns_none_when_empty(self) -> None:
"""Test get_next_download returns None when no eligible downloads."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
result = store.get_next_download()
assert result is None
def test_get_next_download_skips_cancelled(self) -> None:
"""Test get_next_download skips cancelled downloads."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
# Add cancelled download
item1 = make_item(id="vid1")
item1._id = "id1"
store.put(StubDownload(info=item1, started=False, cancelled=True))
# Add eligible download
item2 = make_item(id="vid2")
item2._id = "id2"
store.put(StubDownload(info=item2, started=False, cancelled=False))
result = store.get_next_download()
assert result is not None
assert result.info._id == "id2"
def test_update_store_item_removes_datetime_field(self) -> None:
"""Test that _update_store_item removes datetime field before storage."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
item = make_item(id="vid1")
item.datetime = "Thu, 01 Jan 2024 12:00:00 GMT" # Add datetime field
d = StubDownload(info=item)
store.put(d)
# Verify datetime field is not in stored JSON
row = conn.execute("SELECT data FROM history WHERE id=?", (item._id,)).fetchone()
assert row is not None
data = json.loads(row["data"])
assert "datetime" not in data
def test_update_store_item_removes_live_in_when_finished(self) -> None:
"""Test that _update_store_item removes live_in field when status is finished."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
item = make_item(id="vid1")
item.status = "finished"
item.live_in = "PT5M" # Add live_in field
d = StubDownload(info=item)
store.put(d)
# Verify live_in field is not in stored JSON when status is finished
row = conn.execute("SELECT data FROM history WHERE id=?", (item._id,)).fetchone()
assert row is not None
data = json.loads(row["data"])
assert "live_in" not in data
def test_update_store_item_keeps_live_in_when_not_finished(self) -> None:
"""Test that _update_store_item keeps live_in field when status is not finished."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
item = make_item(id="vid1")
item.status = "downloading"
item.live_in = "PT5M" # Add live_in field
d = StubDownload(info=item)
store.put(d)
# Verify live_in field IS in stored JSON when status is not finished
row = conn.execute("SELECT data FROM history WHERE id=?", (item._id,)).fetchone()
assert row is not None
data = json.loads(row["data"])
assert "live_in" in data
assert data["live_in"] == "PT5M"
class TestDataStoreOperations:
"""Test get_item with different comparison operations."""
def test_operation_equal(self) -> None:
"""Test EQUAL operation (default behavior)."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
item1 = make_item(id="vid1", title="Exact Match", folder="folder1")
item1._id = "id1"
store.put(StubDownload(info=item1))
# Test with explicit EQUAL operation
result = store.get_item(title=(Operation.EQUAL, "Exact Match"))
assert result is not None
assert result.info._id == "id1"
# Test default behavior (no operation specified)
result = store.get_item(title="Exact Match")
assert result is not None
assert result.info._id == "id1"
# Test no match
result = store.get_item(title=(Operation.EQUAL, "No Match"))
assert result is None
def test_operation_not_equal(self) -> None:
"""Test NOT_EQUAL operation."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
item1 = make_item(id="vid1", title="Video 1", folder="folder1")
item1._id = "id1"
item2 = make_item(id="vid2", title="Video 2", folder="folder2")
item2._id = "id2"
store.put(StubDownload(info=item1))
store.put(StubDownload(info=item2))
# Find item where title is not "Video 1"
result = store.get_item(title=(Operation.NOT_EQUAL, "Video 1"))
assert result is not None
assert result.info._id == "id2"
def test_operation_contain(self) -> None:
"""Test CONTAIN operation (substring match)."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
item1 = make_item(id="vid1", title="Python Tutorial Video", folder="folder1")
item1._id = "id1"
item2 = make_item(id="vid2", title="JavaScript Course", folder="folder2")
item2._id = "id2"
store.put(StubDownload(info=item1))
store.put(StubDownload(info=item2))
# Find item with "Python" in title
result = store.get_item(title=(Operation.CONTAIN, "Python"))
assert result is not None
assert result.info._id == "id1"
# Find item with "Tutorial" in title
result = store.get_item(title=(Operation.CONTAIN, "Tutorial"))
assert result is not None
assert result.info._id == "id1"
# No match
result = store.get_item(title=(Operation.CONTAIN, "Rust"))
assert result is None
def test_operation_not_contain(self) -> None:
"""Test NOT_CONTAIN operation."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
item1 = make_item(id="vid1", title="Python Tutorial", folder="folder1")
item1._id = "id1"
item2 = make_item(id="vid2", title="JavaScript Course", folder="folder2")
item2._id = "id2"
store.put(StubDownload(info=item1))
store.put(StubDownload(info=item2))
# Find item that doesn't contain "Python"
result = store.get_item(title=(Operation.NOT_CONTAIN, "Python"))
assert result is not None
assert result.info._id == "id2"
def test_operation_starts_with(self) -> None:
"""Test STARTS_WITH operation."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
item1 = make_item(id="vid1", title="Tutorial: Python Basics", folder="folder1")
item1._id = "id1"
item2 = make_item(id="vid2", title="Course: JavaScript", folder="folder2")
item2._id = "id2"
store.put(StubDownload(info=item1))
store.put(StubDownload(info=item2))
# Find item starting with "Tutorial"
result = store.get_item(title=(Operation.STARTS_WITH, "Tutorial"))
assert result is not None
assert result.info._id == "id1"
# Find item starting with "Course"
result = store.get_item(title=(Operation.STARTS_WITH, "Course"))
assert result is not None
assert result.info._id == "id2"
# No match
result = store.get_item(title=(Operation.STARTS_WITH, "Video"))
assert result is None
def test_operation_ends_with(self) -> None:
"""Test ENDS_WITH operation."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
item1 = make_item(id="vid1", title="Learn Python", folder="folder1")
item1._id = "id1"
item2 = make_item(id="vid2", title="Learn JavaScript", folder="folder2")
item2._id = "id2"
store.put(StubDownload(info=item1))
store.put(StubDownload(info=item2))
# Find item ending with "Python"
result = store.get_item(title=(Operation.ENDS_WITH, "Python"))
assert result is not None
assert result.info._id == "id1"
# Find item ending with "JavaScript"
result = store.get_item(title=(Operation.ENDS_WITH, "JavaScript"))
assert result is not None
assert result.info._id == "id2"
# No match
result = store.get_item(title=(Operation.ENDS_WITH, "Course"))
assert result is None
def test_operation_greater_than(self) -> None:
"""Test GREATER_THAN operation with numeric values."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
item1 = make_item(id="vid1", title="Video 1")
item1._id = "id1"
item1.filesize = 1000
item2 = make_item(id="vid2", title="Video 2")
item2._id = "id2"
item2.filesize = 2000
store.put(StubDownload(info=item1))
store.put(StubDownload(info=item2))
# Find item with filesize > 1500
result = store.get_item(filesize=(Operation.GREATER_THAN, 1500))
assert result is not None
assert result.info._id == "id2"
# Find item with filesize > 500 (should return first match)
result = store.get_item(filesize=(Operation.GREATER_THAN, 500))
assert result is not None
assert result.info._id == "id1"
def test_operation_less_than(self) -> None:
"""Test LESS_THAN operation."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
item1 = make_item(id="vid1", title="Video 1")
item1._id = "id1"
item1.filesize = 1000
item2 = make_item(id="vid2", title="Video 2")
item2._id = "id2"
item2.filesize = 2000
store.put(StubDownload(info=item1))
store.put(StubDownload(info=item2))
# Find item with filesize < 1500
result = store.get_item(filesize=(Operation.LESS_THAN, 1500))
assert result is not None
assert result.info._id == "id1"
def test_operation_greater_equal(self) -> None:
"""Test GREATER_EQUAL operation."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
item1 = make_item(id="vid1", title="Video 1")
item1._id = "id1"
item1.filesize = 1000
store.put(StubDownload(info=item1))
# Test >= with exact match
result = store.get_item(filesize=(Operation.GREATER_EQUAL, 1000))
assert result is not None
assert result.info._id == "id1"
# Test >= with less than
result = store.get_item(filesize=(Operation.GREATER_EQUAL, 500))
assert result is not None
assert result.info._id == "id1"
# Test >= with greater than
result = store.get_item(filesize=(Operation.GREATER_EQUAL, 1500))
assert result is None
def test_operation_less_equal(self) -> None:
"""Test LESS_EQUAL operation."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
item1 = make_item(id="vid1", title="Video 1")
item1._id = "id1"
item1.filesize = 1000
store.put(StubDownload(info=item1))
# Test <= with exact match
result = store.get_item(filesize=(Operation.LESS_EQUAL, 1000))
assert result is not None
# Test <= with greater than
result = store.get_item(filesize=(Operation.LESS_EQUAL, 1500))
assert result is not None
# Test <= with less than
result = store.get_item(filesize=(Operation.LESS_EQUAL, 500))
assert result is None
def test_mixed_operations(self) -> None:
"""Test using multiple operations in a single query."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
item1 = make_item(id="vid1", title="Python Tutorial", folder="tutorials")
item1._id = "id1"
item2 = make_item(id="vid2", title="Python Advanced", folder="courses")
item2._id = "id2"
item3 = make_item(id="vid3", title="JavaScript Basics", folder="tutorials")
item3._id = "id3"
store.put(StubDownload(info=item1))
store.put(StubDownload(info=item2))
store.put(StubDownload(info=item3))
# Mix of operation and default (any match returns true)
result = store.get_item(title=(Operation.CONTAIN, "Python"), folder="tutorials")
assert result is not None
assert result.info._id == "id1"
# Mix where first condition matches
result = store.get_item(title=(Operation.CONTAIN, "JavaScript"), folder="nonexistent")
assert result is not None
assert result.info._id == "id3"
def test_operation_with_none_values(self) -> None:
"""Test operations handle None values gracefully."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
item1 = make_item(id="vid1", title="Video 1")
item1._id = "id1"
item1.description = None
store.put(StubDownload(info=item1))
# CONTAIN with None field should return False
result = store.get_item(description=(Operation.CONTAIN, "test"))
assert result is None
# NOT_CONTAIN with None field should return True
result = store.get_item(description=(Operation.NOT_CONTAIN, "test"))
assert result is not None
# GREATER_THAN with None should return False
result = store.get_item(description=(Operation.GREATER_THAN, 100))
assert result is None
def test_operation_with_invalid_comparisons(self) -> None:
"""Test that invalid comparisons are handled gracefully."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
item1 = make_item(id="vid1", title="Video 1")
item1._id = "id1"
store.put(StubDownload(info=item1))
# Try to compare string with number using > (should return False/None)
result = store.get_item(title=(Operation.GREATER_THAN, 100))
assert result is None
def test_operation_backward_compatibility(self) -> None:
"""Test that string operation names work for backward compatibility."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
item1 = make_item(id="vid1", title="Python Tutorial")
item1._id = "id1"
store.put(StubDownload(info=item1))
# Using string operation value
result = store.get_item(title=("in", "Python"))
assert result is not None
assert result.info._id == "id1"
# Using string for EQUAL
result = store.get_item(title=("==", "Python Tutorial"))
assert result is not None
def test_operation_with_nonexistent_field(self) -> None:
"""Test operations with fields that don't exist."""
conn = make_conn()
store = DataStore(StoreType.QUEUE, conn)
item1 = make_item(id="vid1", title="Video 1")
item1._id = "id1"
store.put(StubDownload(info=item1))
# Try to match on non-existent field
result = store.get_item(nonexistent_field=(Operation.EQUAL, "value"))
assert result is None
result = store.get_item(nonexistent_field=(Operation.CONTAIN, "value"))
assert result is None

244
app/tests/test_operations.py Normal file
View file

@ -0,0 +1,244 @@
"""Tests for the generic operations module."""
from app.library.operations import (
Operation,
filter_items,
find_all,
find_first,
matches,
matches_all,
matches_any,
matches_condition,
)
class TestMatchesGeneric:
"""Test the generic matches(operation, val1, val2) function."""
def test_matches_equal(self) -> None:
"""Test EQUAL operation."""
assert matches(Operation.EQUAL, "test", "test") is True
assert matches(Operation.EQUAL, 100, 100) is True
assert matches(Operation.EQUAL, "test", "other") is False
assert matches(Operation.EQUAL, None, None) is True
def test_matches_not_equal(self) -> None:
"""Test NOT_EQUAL operation."""
assert matches(Operation.NOT_EQUAL, "test", "other") is True
assert matches(Operation.NOT_EQUAL, 100, 200) is True
assert matches(Operation.NOT_EQUAL, "test", "test") is False
def test_matches_contain(self) -> None:
"""Test CONTAIN operation."""
assert matches(Operation.CONTAIN, "Python Tutorial", "Python") is True
assert matches(Operation.CONTAIN, "Hello World", "World") is True
assert matches(Operation.CONTAIN, "test", "xyz") is False
assert matches(Operation.CONTAIN, None, "test") is False
def test_matches_not_contain(self) -> None:
"""Test NOT_CONTAIN operation."""
assert matches(Operation.NOT_CONTAIN, "Python Tutorial", "Java") is True
assert matches(Operation.NOT_CONTAIN, "test", "test") is False
assert matches(Operation.NOT_CONTAIN, None, "test") is True
def test_matches_greater_than(self) -> None:
"""Test GREATER_THAN operation."""
assert matches(Operation.GREATER_THAN, 100, 50) is True
assert matches(Operation.GREATER_THAN, 50, 100) is False
assert matches(Operation.GREATER_THAN, 100, 100) is False
assert matches(Operation.GREATER_THAN, None, 50) is False
assert matches(Operation.GREATER_THAN, 100, None) is False
def test_matches_less_than(self) -> None:
"""Test LESS_THAN operation."""
assert matches(Operation.LESS_THAN, 50, 100) is True
assert matches(Operation.LESS_THAN, 100, 50) is False
assert matches(Operation.LESS_THAN, 100, 100) is False
def test_matches_greater_equal(self) -> None:
"""Test GREATER_EQUAL operation."""
assert matches(Operation.GREATER_EQUAL, 100, 50) is True
assert matches(Operation.GREATER_EQUAL, 100, 100) is True
assert matches(Operation.GREATER_EQUAL, 50, 100) is False
def test_matches_less_equal(self) -> None:
"""Test LESS_EQUAL operation."""
assert matches(Operation.LESS_EQUAL, 50, 100) is True
assert matches(Operation.LESS_EQUAL, 100, 100) is True
assert matches(Operation.LESS_EQUAL, 100, 50) is False
def test_matches_starts_with(self) -> None:
"""Test STARTS_WITH operation."""
assert matches(Operation.STARTS_WITH, "Python Tutorial", "Python") is True
assert matches(Operation.STARTS_WITH, "Tutorial", "Python") is False
assert matches(Operation.STARTS_WITH, None, "test") is False
def test_matches_ends_with(self) -> None:
"""Test ENDS_WITH operation."""
assert matches(Operation.ENDS_WITH, "Learn Python", "Python") is True
assert matches(Operation.ENDS_WITH, "Python Tutorial", "Python") is False
assert matches(Operation.ENDS_WITH, None, "test") is False
def test_matches_with_string_operation(self) -> None:
"""Test backward compatibility with string operations."""
assert matches("==", "test", "test") is True
assert matches("in", "Python Tutorial", "Python") is True
assert matches(">", 100, 50) is True
def test_matches_with_invalid_operation(self) -> None:
"""Test with invalid operation string defaults to EQUAL."""
assert matches("invalid_op", "test", "test") is True
assert matches("invalid_op", "test", "other") is False
def test_matches_with_incompatible_types(self) -> None:
"""Test that incompatible type comparisons return False."""
# Comparing string with number for > operation
assert matches(Operation.GREATER_THAN, "text", 100) is False
class TestMatchesCondition:
"""Test the matches_condition helper function."""
def test_matches_condition_simple_value(self) -> None:
"""Test with simple value (defaults to EQUAL)."""
data = {"title": "Python Tutorial", "size": 1000}
assert matches_condition("title", "Python Tutorial", data) is True
assert matches_condition("title", "Other", data) is False
def test_matches_condition_with_operation(self) -> None:
"""Test with tuple (operation, value)."""
data = {"title": "Python Tutorial", "size": 1000}
assert matches_condition("title", (Operation.CONTAIN, "Python"), data) is True
assert matches_condition("size", (Operation.GREATER_THAN, 500), data) is True
def test_matches_condition_missing_key(self) -> None:
"""Test with non-existent key."""
data = {"title": "Python Tutorial"}
assert matches_condition("missing", "value", data) is False
class TestMatchesAll:
"""Test matches_all function (AND logic)."""
def test_matches_all_true(self) -> None:
"""Test when all conditions match."""
data = {"title": "Python Tutorial", "size": 1000, "status": "active"}
assert matches_all(data, title=(Operation.CONTAIN, "Python"), size=(Operation.GREATER_THAN, 500)) is True
def test_matches_all_false(self) -> None:
"""Test when any condition fails."""
data = {"title": "Python Tutorial", "size": 1000}
assert matches_all(data, title=(Operation.CONTAIN, "Python"), size=(Operation.GREATER_THAN, 2000)) is False
def test_matches_all_empty_conditions(self) -> None:
"""Test with no conditions returns True."""
data = {"title": "Python Tutorial"}
assert matches_all(data) is True
class TestMatchesAny:
"""Test matches_any function (OR logic)."""
def test_matches_any_true(self) -> None:
"""Test when at least one condition matches."""
data = {"title": "Python Tutorial", "size": 1000}
assert matches_any(data, title=(Operation.CONTAIN, "Java"), size=(Operation.GREATER_THAN, 500)) is True
def test_matches_any_false(self) -> None:
"""Test when no conditions match."""
data = {"title": "Python Tutorial", "size": 1000}
assert matches_any(data, title="Wrong", status="Wrong") is False
def test_matches_any_empty_conditions(self) -> None:
"""Test with no conditions returns False."""
data = {"title": "Python Tutorial"}
assert matches_any(data) is False
class TestFilterItems:
"""Test filter_items function."""
def test_filter_items_basic(self) -> None:
"""Test basic filtering."""
items = [
{"title": "Python Tutorial", "size": 1000},
{"title": "JavaScript Course", "size": 2000},
{"title": "Python Advanced", "size": 1500},
]
result = filter_items(items, title=(Operation.CONTAIN, "Python"))
assert len(result) == 2
assert result[0]["title"] == "Python Tutorial"
assert result[1]["title"] == "Python Advanced"
def test_filter_items_multiple_conditions(self) -> None:
"""Test filtering with multiple conditions (AND logic)."""
items = [
{"title": "Python Tutorial", "size": 1000},
{"title": "Python Advanced", "size": 2000},
{"title": "JavaScript Course", "size": 1500},
]
result = filter_items(items, title=(Operation.CONTAIN, "Python"), size=(Operation.GREATER_THAN, 1200))
assert len(result) == 1
assert result[0]["title"] == "Python Advanced"
def test_filter_items_no_conditions(self) -> None:
"""Test that no conditions returns all items."""
items = [{"title": "Test 1"}, {"title": "Test 2"}]
result = filter_items(items)
assert len(result) == 2
class TestFindFirst:
"""Test find_first function."""
def test_find_first_match(self) -> None:
"""Test finding first match."""
items = [
{"title": "Python Tutorial", "size": 1000},
{"title": "Python Advanced", "size": 2000},
]
result = find_first(items, title=(Operation.CONTAIN, "Python"))
assert result is not None
assert result["title"] == "Python Tutorial"
def test_find_first_no_match(self) -> None:
"""Test when no match is found."""
items = [{"title": "Python Tutorial"}]
result = find_first(items, title="Nonexistent")
assert result is None
class TestFindAll:
"""Test find_all function (alias for filter_items)."""
def test_find_all(self) -> None:
"""Test find_all is an alias for filter_items."""
items = [
{"title": "Python Tutorial", "size": 1000},
{"title": "JavaScript Course", "size": 2000},
]
result = find_all(items, title=(Operation.CONTAIN, "Python"))
assert len(result) == 1
assert result[0]["title"] == "Python Tutorial"
class TestOperationEnum:
"""Test Operation enum."""
def test_operation_values(self) -> None:
"""Test operation enum values."""
assert Operation.EQUAL.value == "=="
assert Operation.NOT_EQUAL.value == "!="
assert Operation.CONTAIN.value == "in"
assert Operation.GREATER_THAN.value == ">"
assert Operation.STARTS_WITH.value == "startswith"
def test_operation_str(self) -> None:
"""Test operation string representation."""
assert str(Operation.EQUAL) == "=="
assert str(Operation.CONTAIN) == "in"

333
app/tests/test_utils.py
View file

@ -39,8 +39,10 @@ from app.library.Utils import (
load_cookies,
load_modules,
merge_dict,
move_file,
parse_tags,
read_logfile,
rename_file,
str_to_dt,
strip_newline,
tail_log,
@ -1731,9 +1733,7 @@ class TestGetChannelImages:
"""Test extracting poster image from portrait ratio thumbnail."""
from app.library.Utils import get_channel_images
thumbnails = [
{"url": "http://example.com/poster.jpg", "width": 200, "height": 300, "id": "some_id"}
]
thumbnails = [{"url": "http://example.com/poster.jpg", "width": 200, "height": 300, "id": "some_id"}]
result = get_channel_images(thumbnails)
@ -2049,3 +2049,330 @@ class TestCreateCookiesFile:
assert result == cookie_path
assert cookie_path.read_text() == "new_data"
class TestRenameFile:
"""Test rename_file function."""
def test_rename_single_file_no_sidecars(self, tmp_path: Path):
"""Test renaming a single file without sidecar files."""
# Create test file
test_file = tmp_path / "video.mp4"
test_file.write_text("test content")
# Rename file
new_path, sidecars = rename_file(test_file, "renamed_video.mp4")
# Assertions
assert new_path.exists()
assert "renamed_video.mp4" == new_path.name
assert not test_file.exists()
assert 0 == len(sidecars)
def test_rename_file_with_subtitle_sidecar(self, tmp_path: Path):
"""Test renaming a file with subtitle sidecar."""
# Create test files
test_file = tmp_path / "video.mp4"
test_file.write_text("test video")
subtitle_file = tmp_path / "video.en.srt"
subtitle_file.write_text("test subtitle")
# Rename file
new_path, sidecars = rename_file(test_file, "renamed_video.mp4")
# Assertions
assert new_path.exists()
assert "renamed_video.mp4" == new_path.name
assert not test_file.exists()
assert 1 == len(sidecars)
old_sidecar, new_sidecar = sidecars[0]
assert new_sidecar.exists()
assert "renamed_video.en.srt" == new_sidecar.name
assert old_sidecar == subtitle_file
assert not subtitle_file.exists()
def test_rename_file_with_multiple_sidecars(self, tmp_path: Path):
"""Test renaming a file with multiple sidecar files."""
# Create test files
test_file = tmp_path / "video.mp4"
test_file.write_text("test video")
subtitle_en = tmp_path / "video.en.srt"
subtitle_en.write_text("english subtitle")
subtitle_fr = tmp_path / "video.fr.srt"
subtitle_fr.write_text("french subtitle")
info_file = tmp_path / "video.info.json"
info_file.write_text('{"title": "test"}')
# Rename file
new_path, sidecars = rename_file(test_file, "renamed_video.mp4")
# Assertions
assert new_path.exists()
assert "renamed_video.mp4" == new_path.name
assert not test_file.exists()
assert 3 == len(sidecars)
# Check all sidecars were renamed
sidecar_names = {new_sidecar.name for old_sidecar, new_sidecar in sidecars}
assert "renamed_video.en.srt" in sidecar_names
assert "renamed_video.fr.srt" in sidecar_names
assert "renamed_video.info.json" in sidecar_names
# Check old files don't exist
assert not subtitle_en.exists()
assert not subtitle_fr.exists()
assert not info_file.exists()
def test_rename_file_destination_exists(self, tmp_path: Path):
"""Test renaming a file when destination already exists."""
# Create test files
test_file = tmp_path / "video.mp4"
test_file.write_text("test content")
existing_file = tmp_path / "renamed_video.mp4"
existing_file.write_text("existing content")
# Should raise ValueError
with pytest.raises(ValueError, match="already exists"):
rename_file(test_file, "renamed_video.mp4")
# Original files should still exist
assert test_file.exists()
assert existing_file.exists()
def test_rename_file_sidecar_destination_exists(self, tmp_path: Path):
"""Test renaming when sidecar destination already exists."""
# Create test files
test_file = tmp_path / "video.mp4"
test_file.write_text("test video")
subtitle_file = tmp_path / "video.en.srt"
subtitle_file.write_text("test subtitle")
# Create conflicting sidecar destination
conflicting_sidecar = tmp_path / "renamed_video.en.srt"
conflicting_sidecar.write_text("existing subtitle")
# Should raise ValueError
with pytest.raises(ValueError, match=r"Sidecar destination.*already exists"):
rename_file(test_file, "renamed_video.mp4")
# Original files should still exist
assert test_file.exists()
assert subtitle_file.exists()
assert conflicting_sidecar.exists()
def test_rename_preserves_sidecar_extensions(self, tmp_path: Path):
"""Test that rename preserves complex sidecar extensions."""
# Create test files with complex extensions
test_file = tmp_path / "video.mp4"
test_file.write_text("test video")
subtitle_file = tmp_path / "video.en-US.ass"
subtitle_file.write_text("test subtitle")
thumb_file = tmp_path / "video.thumb.jpg"
thumb_file.write_text("test thumb")
# Rename file
new_path, sidecars = rename_file(test_file, "renamed.mp4")
# Assertions
assert new_path.exists()
assert "renamed.mp4" == new_path.name
assert 2 == len(sidecars)
sidecar_names = {new_sidecar.name for old_sidecar, new_sidecar in sidecars}
assert "renamed.en-US.ass" in sidecar_names
assert "renamed.thumb.jpg" in sidecar_names
class TestMoveFile:
"""Test move_file function."""
def test_move_single_file_no_sidecars(self, tmp_path: Path):
"""Test moving a single file without sidecar files."""
# Create test file
source_dir = tmp_path / "source"
source_dir.mkdir()
test_file = source_dir / "video.mp4"
test_file.write_text("test content")
target_dir = tmp_path / "target"
target_dir.mkdir()
# Move file
new_path, sidecars = move_file(test_file, target_dir)
# Assertions
assert new_path.exists()
assert "video.mp4" == new_path.name
assert new_path.parent == target_dir
assert not test_file.exists()
assert 0 == len(sidecars)
def test_move_file_with_subtitle_sidecar(self, tmp_path: Path):
"""Test moving a file with subtitle sidecar."""
# Create test files
source_dir = tmp_path / "source"
source_dir.mkdir()
test_file = source_dir / "video.mp4"
test_file.write_text("test video")
subtitle_file = source_dir / "video.en.srt"
subtitle_file.write_text("test subtitle")
target_dir = tmp_path / "target"
target_dir.mkdir()
# Move file
new_path, sidecars = move_file(test_file, target_dir)
# Assertions
assert new_path.exists()
assert "video.mp4" == new_path.name
assert new_path.parent == target_dir
assert not test_file.exists()
assert 1 == len(sidecars)
old_sidecar, new_sidecar = sidecars[0]
assert new_sidecar.exists()
assert "video.en.srt" == new_sidecar.name
assert new_sidecar.parent == target_dir
assert old_sidecar == subtitle_file
assert not subtitle_file.exists()
def test_move_file_with_multiple_sidecars(self, tmp_path: Path):
"""Test moving a file with multiple sidecar files."""
# Create test files
source_dir = tmp_path / "source"
source_dir.mkdir()
test_file = source_dir / "video.mp4"
test_file.write_text("test video")
subtitle_en = source_dir / "video.en.srt"
subtitle_en.write_text("english subtitle")
subtitle_fr = source_dir / "video.fr.srt"
subtitle_fr.write_text("french subtitle")
info_file = source_dir / "video.info.json"
info_file.write_text('{"title": "test"}')
target_dir = tmp_path / "target"
target_dir.mkdir()
# Move file
new_path, sidecars = move_file(test_file, target_dir)
# Assertions
assert new_path.exists()
assert "video.mp4" == new_path.name
assert new_path.parent == target_dir
assert not test_file.exists()
assert 3 == len(sidecars)
# Check all sidecars were moved
sidecar_names = {new_sidecar.name for old_sidecar, new_sidecar in sidecars}
assert "video.en.srt" in sidecar_names
assert "video.fr.srt" in sidecar_names
assert "video.info.json" in sidecar_names
# Check all are in target directory
for _old_sidecar, new_sidecar in sidecars:
assert new_sidecar.parent == target_dir
# Check old files don't exist
assert not subtitle_en.exists()
assert not subtitle_fr.exists()
assert not info_file.exists()
def test_move_file_destination_exists(self, tmp_path: Path):
"""Test moving a file when destination already exists."""
# Create test files
source_dir = tmp_path / "source"
source_dir.mkdir()
test_file = source_dir / "video.mp4"
test_file.write_text("test content")
target_dir = tmp_path / "target"
target_dir.mkdir()
existing_file = target_dir / "video.mp4"
existing_file.write_text("existing content")
# Should raise ValueError
with pytest.raises(ValueError, match="already exists"):
move_file(test_file, target_dir)
# Original files should still exist
assert test_file.exists()
assert existing_file.exists()
def test_move_file_sidecar_destination_exists(self, tmp_path: Path):
"""Test moving when sidecar destination already exists."""
# Create test files
source_dir = tmp_path / "source"
source_dir.mkdir()
test_file = source_dir / "video.mp4"
test_file.write_text("test video")
subtitle_file = source_dir / "video.en.srt"
subtitle_file.write_text("test subtitle")
target_dir = tmp_path / "target"
target_dir.mkdir()
# Create conflicting sidecar destination
conflicting_sidecar = target_dir / "video.en.srt"
conflicting_sidecar.write_text("existing subtitle")
# Should raise ValueError
with pytest.raises(ValueError, match=r"Sidecar destination.*already exists"):
move_file(test_file, target_dir)
# Original files should still exist
assert test_file.exists()
assert subtitle_file.exists()
assert conflicting_sidecar.exists()
def test_move_file_target_not_directory(self, tmp_path: Path):
"""Test moving when target is not a directory."""
# Create test file
source_dir = tmp_path / "source"
source_dir.mkdir()
test_file = source_dir / "video.mp4"
test_file.write_text("test content")
# Create a file (not directory) as target
target_file = tmp_path / "target.txt"
target_file.write_text("not a directory")
# Should raise ValueError
with pytest.raises(ValueError, match="not a directory"):
move_file(test_file, target_file)
# Original file should still exist
assert test_file.exists()
def test_move_file_target_does_not_exist(self, tmp_path: Path):
"""Test moving when target directory doesn't exist."""
# Create test file
source_dir = tmp_path / "source"
source_dir.mkdir()
test_file = source_dir / "video.mp4"
test_file.write_text("test content")
target_dir = tmp_path / "nonexistent"
# Should raise ValueError
with pytest.raises(ValueError, match="does not exist"):
move_file(test_file, target_dir)
# Original file should still exist
assert test_file.exists()

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

@ -51,13 +51,6 @@ const sendMessage = (type: notificationType, id: string, message: string, opts?:
const useToastNotification = !window.isSecureContext || 'toast' === toastTarget.value ||
!('Notification' in window) || 'granted' !== Notification.permission;
console.log('useToastNotification', useToastNotification,{
windowIsSecureContext: window.isSecureContext,
notificationTarget: toastTarget.value,
notificationInWindow: 'Notification' in window,
notificationPermission: Notification.permission
});
if (useToastNotification) {
switch (type) {
case 'info':

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

@ -726,9 +726,10 @@ const handleAction = async (action: string, item: FileItem): Promise<void> => {
}
if ('rename' === action) {
const moveSideCars = 'file' === item.type ? ' (and its sidecars)' : ''
const { status, value: newName } = await dialog.promptDialog({
title: 'Rename Item',
message: `Enter new name for '${item.name}':`,
message: `Enter new name for '${item.name}'${moveSideCars}:`,
initial: item.name,
confirmText: 'Rename',
cancelText: 'Cancel',
@ -742,11 +743,13 @@ const handleAction = async (action: string, item: FileItem): Promise<void> => {
return
}
await actionRequest(item, 'rename', { new_name: new_name }, (item, _, data) => {
item.name = data.new_name
item.path = item.path.replace(/[^/]+$/, data.new_name)
toast.success(`Renamed '${item.name}'.`)
})
await actionRequest(item, 'rename', { new_name: new_name }, (item, _, data, source) => {
if (item.path === source.path) {
toast.success(`Renamed '${item.name}'.`)
}
item.name = basename(data.new_path)
item.path = data.new_path
}, true)
return
}
@ -773,9 +776,10 @@ const handleAction = async (action: string, item: FileItem): Promise<void> => {
}
if ('move' === action) {
const moveSideCars = 'file' === item.type ? ' (and its sidecars)' : ''
const { status, value: newPath } = await dialog.promptDialog({
title: 'Move Item',
message: `Enter new path for '${item.name}':`,
message: `Enter new path for '${item.name}'${moveSideCars}:`,
initial: item.path.replace(/[^/]+$/, '') || '/',
confirmText: 'Move',
cancelText: 'Cancel',
@ -790,10 +794,12 @@ const handleAction = async (action: string, item: FileItem): Promise<void> => {
return
}
await actionRequest(item, 'move', { new_path: new_path }, (item, action, data) => {
items.value = items.value.filter(i => i.path !== item.path)
toast.success(`Moved '${item.name}' to '${data.new_path}'.`)
})
await actionRequest(item, 'move', { new_path: new_path }, (item, _, data, source) => {
if (item.path === source.path) {
toast.success(`Moved '${item.name}' to '${data.new_path}'.`)
}
items.value = items.value.filter(i => i.path !== data.path)
}, true)
return
}
@ -802,25 +808,22 @@ const handleAction = async (action: string, item: FileItem): Promise<void> => {
const actionRequest = async (
item: FileItem,
action: string,
data: Record<string, any>,
cb: (item: FileItem, action: string, data: any) => void
req: Record<string, any>,
cb: (item: FileItem, action: string, data: any, source: FileItem, req: any) => void,
multiple: boolean = false
): Promise<any> => {
if (!config.app.browser_control_enabled) {
return
}
if (!item || !action || !data) {
if (!item || !action || !req) {
return
}
try {
const response = await request(`/api/file/actions`, {
method: 'POST',
body: JSON.stringify([{
path: item.path,
action: action,
...data
}]),
body: JSON.stringify([{ path: item.path, action: action, ...req }]),
})
if (!response.ok) {
@ -831,17 +834,24 @@ const actionRequest = async (
const json = await response.json() as Array<{ path: string, status: boolean, error?: string }>
json.forEach(i => {
if (i.path !== item.path) {
if (false === multiple && i.path !== item.path) {
return
}
if (true !== i.status) {
if (false === multiple && true !== i.status) {
toast.error(`Failed to perform action: ${i.error || 'Unknown error'}`)
return
}
if (cb && typeof cb === 'function') {
cb(item, action, data)
if (false === multiple) {
return cb(item, action, req, item, req)
}
items.value.forEach(it => {
if (it.path === i.path) {
return cb(it, action, i, toRaw(item), req)
}
})
}
});