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"