From 6d9c7d70d9e5c496d777d981cb5d961262c4d890 Mon Sep 17 00:00:00 2001 From: arabcoders Date: Fri, 14 Nov 2025 17:18:16 +0300 Subject: [PATCH 1/3] Feat: update datastore.get_items to support operations --- app/library/DataStore.py | 30 +- app/library/operations.py | 275 ++++++++++++ app/tests/test_datastore.py | 838 ++++++++++++++++++++++++++++++++++- app/tests/test_operations.py | 244 ++++++++++ 4 files changed, 1377 insertions(+), 10 deletions(-) create mode 100644 app/library/operations.py create mode 100644 app/tests/test_operations.py diff --git a/app/library/DataStore.py b/app/library/DataStore.py index 193425a1..c4bdb581 100644 --- a/app/library/DataStore.py +++ b/app/library/DataStore.py @@ -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: diff --git a/app/library/operations.py b/app/library/operations.py new file mode 100644 index 00000000..0a6c0dbf --- /dev/null +++ b/app/library/operations.py @@ -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) + diff --git a/app/tests/test_datastore.py b/app/tests/test_datastore.py index acfcc09e..d7f8a717 100644 --- a/app/tests/test_datastore.py +++ b/app/tests/test_datastore.py @@ -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 diff --git a/app/tests/test_operations.py b/app/tests/test_operations.py new file mode 100644 index 00000000..b0f5f88d --- /dev/null +++ b/app/tests/test_operations.py @@ -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" From 1c7444e2322b3f461f28c8e2286a9c642075a77c Mon Sep 17 00:00:00 2001 From: arabcoders Date: Fri, 14 Nov 2025 18:30:12 +0300 Subject: [PATCH 2/3] Feat: Renaming files in file browser should trigger item update if exists and update sidecars as well. --- app/library/Utils.py | 61 +++++++++++ app/routes/api/browser.py | 45 +++++++- app/tests/test_utils.py | 146 +++++++++++++++++++++++++- ui/app/composables/useNotification.ts | 7 -- ui/app/pages/browser/[...slug].vue | 42 ++++---- 5 files changed, 268 insertions(+), 33 deletions(-) diff --git a/app/library/Utils.py b/app/library/Utils.py index 1313f99f..d534299e 100644 --- a/app/library/Utils.py +++ b/app/library/Utils.py @@ -763,6 +763,67 @@ 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 get_possible_images(dir: str) -> list[dict]: images: list = [] diff --git a/app/routes/api/browser.py b/app/routes/api/browser.py index 2c47962c..e0626974 100644 --- a/app/routes/api/browser.py +++ b/app/routes/api/browser.py @@ -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, 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,47 @@ async def path_actions(request: Request, config: Config) -> Response: continue try: - renamed = path.rename(new_path) + # Rename main file and all sidecar files + renamed, renamed_sidecars = rename_file(path, new_name) + + sidecar_count: int = len(renamed_sidecars) + 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 renamed_sidecars: + extra_info["sidecar_count"] = len(renamed_sidecars) + record(path, ok=True, action=action, extra=extra_info) + + for old_sidecar, new_sidecar in renamed_sidecars: + record( + old_sidecar, + ok=True, + action="rename_sidecar", + extra={"new_path": new_sidecar, "parent": str(path.relative_to(config.download_path))}, + ) + + 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 len(renamed_sidecars) > 0: + item.info.get_file_sidecar() + + queue.done.put(item) + notify.emit(Events.ITEM_UPDATED, data=item.info) if "delete" == action: try: diff --git a/app/tests/test_utils.py b/app/tests/test_utils.py index 536b8ca8..f41feda4 100644 --- a/app/tests/test_utils.py +++ b/app/tests/test_utils.py @@ -41,6 +41,7 @@ from app.library.Utils import ( merge_dict, parse_tags, read_logfile, + rename_file, str_to_dt, strip_newline, tail_log, @@ -1731,9 +1732,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 +2048,144 @@ 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 diff --git a/ui/app/composables/useNotification.ts b/ui/app/composables/useNotification.ts index e9d5cc5d..7d54820b 100644 --- a/ui/app/composables/useNotification.ts +++ b/ui/app/composables/useNotification.ts @@ -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': diff --git a/ui/app/pages/browser/[...slug].vue b/ui/app/pages/browser/[...slug].vue index f089eefa..f3e30990 100644 --- a/ui/app/pages/browser/[...slug].vue +++ b/ui/app/pages/browser/[...slug].vue @@ -728,7 +728,7 @@ const handleAction = async (action: string, item: FileItem): Promise => { if ('rename' === action) { const { status, value: newName } = await dialog.promptDialog({ title: 'Rename Item', - message: `Enter new name for '${item.name}':`, + message: `Enter new name for '${item.name}' Sidecars will be renamed as well:`, initial: item.name, confirmText: 'Rename', cancelText: 'Cancel', @@ -742,11 +742,13 @@ const handleAction = async (action: string, item: FileItem): Promise => { 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 } @@ -790,7 +792,7 @@ const handleAction = async (action: string, item: FileItem): Promise => { return } - await actionRequest(item, 'move', { new_path: new_path }, (item, action, data) => { + await actionRequest(item, 'move', { new_path: new_path }, (item, _, data) => { items.value = items.value.filter(i => i.path !== item.path) toast.success(`Moved '${item.name}' to '${data.new_path}'.`) }) @@ -802,25 +804,22 @@ const handleAction = async (action: string, item: FileItem): Promise => { const actionRequest = async ( item: FileItem, action: string, - data: Record, - cb: (item: FileItem, action: string, data: any) => void + req: Record, + cb: (item: FileItem, action: string, data: any, source: FileItem, req: any) => void, + multiple: boolean = false ): Promise => { 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 +830,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) + } + }) } }); From 4a69b0b6a1fdfd445c50fb5e97155c5d70d2085d Mon Sep 17 00:00:00 2001 From: arabcoders Date: Fri, 14 Nov 2025 20:05:58 +0300 Subject: [PATCH 3/3] Feat: Moving files in file browser should trigger item update if exists and update sidecars as well. --- app/library/Utils.py | 73 +++++++++++ app/routes/api/browser.py | 84 +++++++++---- app/tests/test_utils.py | 187 +++++++++++++++++++++++++++++ ui/app/pages/browser/[...slug].vue | 16 ++- 4 files changed, 329 insertions(+), 31 deletions(-) diff --git a/app/library/Utils.py b/app/library/Utils.py index d534299e..d815a2b8 100644 --- a/app/library/Utils.py +++ b/app/library/Utils.py @@ -824,6 +824,79 @@ def rename_file(old_path: Path, new_name: str) -> tuple[Path, list[tuple[Path, P 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 = [] diff --git a/app/routes/api/browser.py b/app/routes/api/browser.py index e0626974..4066f0dd 100644 --- a/app/routes/api/browser.py +++ b/app/routes/api/browser.py @@ -14,7 +14,7 @@ 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, rename_file +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__) @@ -364,15 +364,19 @@ async def path_actions(request: Request, config: Config, queue: DownloadQueue, n continue try: - # Rename main file and all sidecar files - renamed, renamed_sidecars = rename_file(path, new_name) - - sidecar_count: int = len(renamed_sidecars) - sidecar_info: str = ( - f" (with {sidecar_count} sidecar file{'s' if sidecar_count != 1 else ''})" - if sidecar_count > 0 - else "" - ) + 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)}'{sidecar_info}" @@ -386,21 +390,16 @@ async def path_actions(request: Request, config: Config, queue: DownloadQueue, n continue else: extra_info: dict[str, Path] = {"new_path": renamed} - if renamed_sidecars: - extra_info["sidecar_count"] = len(renamed_sidecars) + 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 renamed_sidecars: - record( - old_sidecar, - ok=True, - action="rename_sidecar", - extra={"new_path": new_sidecar, "parent": str(path.relative_to(config.download_path))}, - ) + 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 len(renamed_sidecars) > 0: + if sidecar_renamed: item.info.get_file_sidecar() queue.done.put(item) @@ -432,7 +431,7 @@ async def path_actions(request: Request, config: Config, queue: DownloadQueue, n 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("/")) @@ -446,7 +445,7 @@ async def path_actions(request: Request, config: Config, queue: DownloadQueue, n ) 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} @@ -468,14 +467,49 @@ async def path_actions(request: Request, config: Config, queue: DownloadQueue, n 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) diff --git a/app/tests/test_utils.py b/app/tests/test_utils.py index f41feda4..5b4893dd 100644 --- a/app/tests/test_utils.py +++ b/app/tests/test_utils.py @@ -39,6 +39,7 @@ from app.library.Utils import ( load_cookies, load_modules, merge_dict, + move_file, parse_tags, read_logfile, rename_file, @@ -2189,3 +2190,189 @@ class TestRenameFile: 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() diff --git a/ui/app/pages/browser/[...slug].vue b/ui/app/pages/browser/[...slug].vue index f3e30990..3cb6e827 100644 --- a/ui/app/pages/browser/[...slug].vue +++ b/ui/app/pages/browser/[...slug].vue @@ -726,9 +726,10 @@ const handleAction = async (action: string, item: FileItem): Promise => { } 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}' Sidecars will be renamed as well:`, + message: `Enter new name for '${item.name}'${moveSideCars}:`, initial: item.name, confirmText: 'Rename', cancelText: 'Cancel', @@ -775,9 +776,10 @@ const handleAction = async (action: string, item: FileItem): Promise => { } 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', @@ -792,10 +794,12 @@ const handleAction = async (action: string, item: FileItem): Promise => { return } - await actionRequest(item, 'move', { new_path: new_path }, (item, _, 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 }