import json from datetime import UTC, datetime import pytest import pytest_asyncio from app.library.DataStore import DataStore, StoreType from app.library.downloads import Download from app.library.ItemDTO import ItemDTO from app.library.sqlite_store import SqliteStore from app.tests.helpers import make_in_memory_db_path async def reset_sqlite_store() -> None: """Close and reset SqliteStore singleton for testing.""" if SqliteStore in SqliteStore._instances: instance = SqliteStore._instances[SqliteStore] # Only close if there's an active connection to avoid event loop issues if instance._conn is not None: try: await instance.close() except RuntimeError: # Event loop issues - just reset without closing pass SqliteStore._reset_singleton() async def make_db(data: int = 100) -> SqliteStore: """Create a named in-memory database with test data.""" await reset_sqlite_store() db_path = make_in_memory_db_path("test-datastore-pagination") ins = SqliteStore.get_instance(db_path=db_path) await ins.get_connection() base_time = datetime.now(UTC) for i in range(data): created_at = base_time.replace(hour=(i // 4) % 24, minute=(i * 15) % 60, second=i % 60).strftime( "%Y-%m-%d %H:%M:%S" ) item_data = { "url": f"https://example.com/video{i}", "title": f"Test Video {i}", "id": f"video{i}", "folder": "/downloads", "status": "finished", } await ins.execute_raw( 'INSERT INTO "history" ("id", "type", "url", "data", "created_at") VALUES (?, ?, ?, ?, ?)', ( f"test-id-{i}", str(StoreType.HISTORY), item_data["url"], json.dumps(item_data), created_at, ), ) return ins @pytest.mark.asyncio class TestDataStorePagination: """Test pagination functionality of DataStore.""" @pytest_asyncio.fixture async def make_db(self, data: int = 100): """Fixture to provide a temporary database with test data.""" return await make_db(data=data) async def test_get_total_count(self): """Test getting total count of items.""" db = await make_db(data=100) try: datastore = DataStore(type=StoreType.HISTORY, connection=db) count = await datastore.get_total_count() assert count == 100 finally: await db.close() async def test_pagination_basic(self): """Test basic pagination functionality.""" db = await make_db(data=100) try: datastore = DataStore(type=StoreType.HISTORY, connection=db) items, total, page, total_pages = await datastore.get_items_paginated(page=1, per_page=10) assert len(items) == 10 assert total == 100 assert page == 1 assert total_pages == 10 for _, item in items: assert isinstance(item, Download) assert isinstance(item.info, ItemDTO) assert item.info._id.startswith("test-id-") finally: await db.close() async def test_pagination_last_page(self): """Test pagination on last page.""" db = await make_db(data=100) try: datastore = DataStore(type=StoreType.HISTORY, connection=db) items, total, page, total_pages = await datastore.get_items_paginated(page=10, per_page=10) assert len(items) == 10 assert total == 100 assert page == 10 assert total_pages == 10 finally: await db.close() async def test_pagination_partial_page(self): """Test pagination with partial last page.""" db = await make_db(data=100) try: datastore = DataStore(type=StoreType.HISTORY, connection=db) items, total, page, total_pages = await datastore.get_items_paginated(page=4, per_page=30) assert len(items) == 10 # 100 items / 30 per page = 3 full pages + 10 items assert total == 100 assert page == 4 assert total_pages == 4 finally: await db.close() async def test_pagination_out_of_range(self): """Test pagination with page number out of range.""" db = await make_db(data=100) try: datastore = DataStore(type=StoreType.HISTORY, connection=db) items, total, page, total_pages = await datastore.get_items_paginated(page=999, per_page=10) assert len(items) == 10 assert total == 100 assert page == 10 # Adjusted to last page assert total_pages == 10 finally: await db.close() async def test_pagination_order_desc(self): """Test pagination with descending order (newest first).""" db = await make_db(data=100) try: datastore = DataStore(type=StoreType.HISTORY, connection=db) items, _, _, _ = await datastore.get_items_paginated(page=1, per_page=5, order="DESC") assert len(items) == 5 finally: await db.close() async def test_pagination_order_asc(self): """Test pagination with ascending order (oldest first).""" db = await make_db(data=100) try: datastore = DataStore(type=StoreType.HISTORY, connection=db) items, _, _, _ = await datastore.get_items_paginated(page=1, per_page=5, order="ASC") assert len(items) == 5 finally: await db.close() async def test_pagination_invalid_page(self): """Test pagination with invalid page number.""" db = await make_db(data=100) try: datastore = DataStore(type=StoreType.HISTORY, connection=db) with pytest.raises(ValueError, match="page must be >= 1"): await datastore.get_items_paginated(page=0, per_page=10) with pytest.raises(ValueError, match="page must be >= 1"): await datastore.get_items_paginated(page=-1, per_page=10) finally: await db.close() async def test_pagination_invalid_per_page(self): """Test pagination with invalid per_page value.""" db = await make_db(data=100) try: datastore = DataStore(type=StoreType.HISTORY, connection=db) with pytest.raises(ValueError, match="per_page must be >= 1"): await datastore.get_items_paginated(page=1, per_page=0) with pytest.raises(ValueError, match="per_page must be >= 1"): await datastore.get_items_paginated(page=1, per_page=-10) finally: await db.close() async def test_pagination_invalid_order(self): """Test pagination with invalid order parameter.""" db = await make_db(data=100) try: datastore = DataStore(type=StoreType.HISTORY, connection=db) with pytest.raises(ValueError, match="order must be 'ASC' or 'DESC'"): await datastore.get_items_paginated(page=1, per_page=10, order="INVALID") finally: await db.close() async def test_pagination_empty_store(self): """Test pagination with empty datastore.""" db = await make_db(data=0) try: datastore = DataStore(type=StoreType.HISTORY, connection=db) items, total, page, total_pages = await datastore.get_items_paginated(page=1, per_page=10) assert len(items) == 0 assert total == 0 assert page == 1 assert total_pages == 1 finally: await db.close() async def test_pagination_single_item(self): """Test pagination with single item.""" db = await make_db(data=1) try: datastore = DataStore(type=StoreType.HISTORY, connection=db) items, total, page, total_pages = await datastore.get_items_paginated(page=1, per_page=10) assert len(items) == 1 assert total == 1 assert page == 1 assert total_pages == 1 finally: await db.close() async def test_pagination_status_filter_include(self): """Test pagination with status filter (inclusion).""" # Add some items with different status values item_data_pending = { "url": "https://example.com/pending", "title": "Pending Video", "id": "pending-video", "folder": "/downloads", "status": "pending", } item_data_downloading = { "url": "https://example.com/downloading", "title": "Downloading Video", "id": "downloading-video", "folder": "/downloads", "status": "downloading", } item_data_skip = { "url": "https://example.com/skip", "title": "Skipped Video", "id": "skip-video", "folder": "/downloads", "status": "skip", } db = await make_db(data=100) try: await db.execute_raw( 'INSERT INTO "history" ("id", "type", "url", "data", "created_at") VALUES (?, ?, ?, ?, ?)', ( "pending-id", str(StoreType.HISTORY), item_data_pending["url"], json.dumps(item_data_pending), datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S"), ), ) await db.execute_raw( 'INSERT INTO "history" ("id", "type", "url", "data", "created_at") VALUES (?, ?, ?, ?, ?)', ( "downloading-id", str(StoreType.HISTORY), item_data_downloading["url"], json.dumps(item_data_downloading), datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S"), ), ) await db.execute_raw( 'INSERT INTO "history" ("id", "type", "url", "data", "created_at") VALUES (?, ?, ?, ?, ?)', ( "skip-id", str(StoreType.HISTORY), item_data_skip["url"], json.dumps(item_data_skip), datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S"), ), ) datastore = DataStore(type=StoreType.HISTORY, connection=db) # Filter for finished items only items, total, _, __ = await datastore.get_items_paginated(page=1, per_page=50, status_filter="finished") assert len(items) == 50 # First page of finished items assert total == 100 # Only 100 finished items in fixture for _item_id, item in items: assert item.info.status == "finished" # Filter for pending items only items, total, _page, _total_pages = await datastore.get_items_paginated( page=1, per_page=50, status_filter="pending" ) assert len(items) == 1 assert total == 1 assert items[0][1].info.status == "pending" items, total, _page, _total_pages = await datastore.get_items_paginated( page=1, per_page=200, status_filter="finished,skip" ) assert total == 101 assert len(items) == 101 assert {item.info.status for _, item in items} == {"finished", "skip"} finally: await db.close() async def test_pagination_status_filter_exclude(self): """Test pagination with status filter (exclusion).""" # Add some items with different status values item_data_pending = { "url": "https://example.com/pending2", "title": "Pending Video 2", "id": "pending-video-2", "folder": "/downloads", "status": "pending", } item_data_error = { "url": "https://example.com/error", "title": "Error Video", "id": "error-video", "folder": "/downloads", "status": "error", } item_data_skip = { "url": "https://example.com/skip2", "title": "Skipped Video 2", "id": "skip-video-2", "folder": "/downloads", "status": "skip", } db = await make_db(data=0) try: await db.execute_raw( 'INSERT INTO "history" ("id", "type", "url", "data", "created_at") VALUES (?, ?, ?, ?, ?)', ( "pending-id-2", str(StoreType.HISTORY), item_data_pending["url"], json.dumps(item_data_pending), datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S"), ), ) await db.execute_raw( 'INSERT INTO "history" ("id", "type", "url", "data", "created_at") VALUES (?, ?, ?, ?, ?)', ( "error-id", str(StoreType.HISTORY), item_data_error["url"], json.dumps(item_data_error), datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S"), ), ) await db.execute_raw( 'INSERT INTO "history" ("id", "type", "url", "data", "created_at") VALUES (?, ?, ?, ?, ?)', ( "skip-id-2", str(StoreType.HISTORY), item_data_skip["url"], json.dumps(item_data_skip), datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S"), ), ) datastore = DataStore(type=StoreType.HISTORY, connection=db) # Exclude finished items items, total, _page, _total_pages = await datastore.get_items_paginated( page=1, per_page=50, status_filter="!finished" ) assert total == 3 assert len(items) == 3 for _item_id, item in items: assert item.info.status != "finished" # Verify we have pending, error, and skip statuses = {item.info.status for _, item in items} assert statuses == {"pending", "error", "skip"} items, total, _page, _total_pages = await datastore.get_items_paginated( page=1, per_page=50, status_filter="!finished,!skip" ) assert total == 2 assert len(items) == 2 assert {item.info.status for _, item in items} == {"pending", "error"} finally: await db.close() async def test_pagination_status_filter_none_matching(self): """Test pagination with status filter that matches no items.""" db = await make_db(data=0) try: datastore = DataStore(type=StoreType.HISTORY, connection=db) # Filter for a status that doesn't exist items, total, page, total_pages = await datastore.get_items_paginated( page=1, per_page=50, status_filter="nonexistent" ) assert len(items) == 0 assert total == 0 assert page == 1 assert total_pages == 1 finally: await db.close()