""" Extended tests for database/queue_service.py - UserQueueService class. Tests cover: - _safe_commit rollback behavior on exceptions - update_queue_status creating vs updating QueueStatus - get_queue_status returning None when no status exists - add_task_metadata creating TaskMetadata and incrementing queue count - update_task_status transitions (queued->processing, processing->completed, processing->failed) and their side effects on timestamps and queue counts - update_task_status handling of non-existent tasks - get_pending_tasks returning ordered list of task dicts - cleanup_old_tasks deleting old completed/failed tasks - _update_queue_counts preventing negative values via max(0, ...) """ from datetime import UTC, datetime from unittest.mock import Mock, patch import pytest class TestSafeCommit: """Tests for _safe_commit rollback behavior.""" def test_safe_commit_calls_session_commit(self): """Successful commit should just call session.commit().""" from local_deep_research.database.queue_service import UserQueueService mock_session = Mock() service = UserQueueService(mock_session) service._safe_commit() mock_session.commit.assert_called_once() mock_session.rollback.assert_not_called() def test_safe_commit_rolls_back_on_exception(self): """On commit failure, should rollback and re-raise.""" from local_deep_research.database.queue_service import UserQueueService mock_session = Mock() mock_session.commit.side_effect = RuntimeError("DB write failed") service = UserQueueService(mock_session) with pytest.raises(RuntimeError, match="DB write failed"): service._safe_commit() mock_session.rollback.assert_called_once() def test_safe_commit_logs_exception_on_failure(self): """On commit failure, should log the exception via loguru.""" from local_deep_research.database.queue_service import UserQueueService mock_session = Mock() mock_session.commit.side_effect = Exception("some error") service = UserQueueService(mock_session) with patch( "local_deep_research.database.queue_service.logger" ) as mock_logger: with pytest.raises(Exception): service._safe_commit() mock_logger.exception.assert_called_once() class TestUpdateQueueStatusExtended: """Extended tests for update_queue_status method.""" def test_creates_new_queue_status_when_none_exists(self): """When no QueueStatus row exists, should create one and add it.""" from local_deep_research.database.queue_service import UserQueueService mock_session = Mock() mock_session.query.return_value.first.return_value = None service = UserQueueService(mock_session) with patch( "local_deep_research.database.queue_service.QueueStatus" ) as MockQueueStatus: mock_new_status = Mock() MockQueueStatus.return_value = mock_new_status service.update_queue_status(3, 7, "task-abc") MockQueueStatus.assert_called_once_with( active_tasks=3, queued_tasks=7, last_task_id="task-abc", ) mock_session.add.assert_called_once_with(mock_new_status) mock_session.commit.assert_called_once() def test_updates_existing_queue_status(self): """When QueueStatus exists, should update its fields in place.""" from local_deep_research.database.queue_service import UserQueueService mock_session = Mock() mock_status = Mock() mock_status.active_tasks = 1 mock_status.queued_tasks = 2 mock_status.last_task_id = "old-task" mock_session.query.return_value.first.return_value = mock_status service = UserQueueService(mock_session) service.update_queue_status(10, 20, "new-task") assert mock_status.active_tasks == 10 assert mock_status.queued_tasks == 20 assert mock_status.last_task_id == "new-task" assert mock_status.last_checked is not None mock_session.commit.assert_called_once() # Should NOT call session.add since status already exists mock_session.add.assert_not_called() def test_does_not_update_last_task_id_when_none(self): """When last_task_id is None, should not overwrite existing value.""" from local_deep_research.database.queue_service import UserQueueService mock_session = Mock() mock_status = Mock() mock_status.active_tasks = 0 mock_status.queued_tasks = 0 mock_status.last_task_id = "keep-this" mock_session.query.return_value.first.return_value = mock_status service = UserQueueService(mock_session) service.update_queue_status(1, 1, last_task_id=None) # last_task_id should remain unchanged because the if-branch is falsy assert mock_status.last_task_id == "keep-this" def test_update_queue_status_sets_last_checked_timestamp(self): """Updating existing status should set last_checked to current time.""" from local_deep_research.database.queue_service import UserQueueService mock_session = Mock() mock_status = Mock() mock_status.last_checked = None mock_session.query.return_value.first.return_value = mock_status service = UserQueueService(mock_session) service.update_queue_status(0, 0) # last_checked should be set to a datetime assert mock_status.last_checked is not None class TestGetQueueStatusExtended: """Extended tests for get_queue_status method.""" def test_returns_none_when_no_status(self): """Should return None when no QueueStatus exists.""" from local_deep_research.database.queue_service import UserQueueService mock_session = Mock() mock_session.query.return_value.first.return_value = None service = UserQueueService(mock_session) result = service.get_queue_status() assert result is None def test_returns_complete_dict_with_all_fields(self): """Should return dict with active_tasks, queued_tasks, last_checked, last_task_id.""" from local_deep_research.database.queue_service import UserQueueService mock_session = Mock() now = datetime.now(UTC) mock_status = Mock() mock_status.active_tasks = 5 mock_status.queued_tasks = 12 mock_status.last_checked = now mock_status.last_task_id = "task-xyz" mock_session.query.return_value.first.return_value = mock_status service = UserQueueService(mock_session) result = service.get_queue_status() assert result == { "active_tasks": 5, "queued_tasks": 12, "last_checked": now, "last_task_id": "task-xyz", } def test_returns_dict_with_null_last_task_id(self): """Should handle None last_task_id correctly.""" from local_deep_research.database.queue_service import UserQueueService mock_session = Mock() mock_status = Mock() mock_status.active_tasks = 0 mock_status.queued_tasks = 0 mock_status.last_checked = datetime.now(UTC) mock_status.last_task_id = None mock_session.query.return_value.first.return_value = mock_status service = UserQueueService(mock_session) result = service.get_queue_status() assert result is not None assert result["last_task_id"] is None class TestAddTaskMetadataExtended: """Extended tests for add_task_metadata method.""" def test_creates_task_metadata_and_increments_queue(self): """Should create TaskMetadata with correct fields and increment queue count.""" from local_deep_research.database.queue_service import UserQueueService mock_session = Mock() mock_status = Mock() mock_status.queued_tasks = 3 mock_session.query.return_value.first.return_value = mock_status service = UserQueueService(mock_session) with patch( "local_deep_research.database.queue_service.TaskMetadata" ) as MockTaskMetadata: mock_task = Mock() MockTaskMetadata.return_value = mock_task service.add_task_metadata("task-99", "research", priority=10) MockTaskMetadata.assert_called_once_with( task_id="task-99", status="queued", task_type="research", priority=10, ) # session.add should be called for the task (and possibly for status) mock_session.add.assert_any_call(mock_task) # queued_tasks should have been incremented assert mock_status.queued_tasks == 4 mock_session.commit.assert_called_once() def test_default_priority_is_zero(self): """Should use priority=0 when not specified.""" from local_deep_research.database.queue_service import UserQueueService mock_session = Mock() mock_status = Mock() mock_status.queued_tasks = 0 mock_session.query.return_value.first.return_value = mock_status service = UserQueueService(mock_session) with patch( "local_deep_research.database.queue_service.TaskMetadata" ) as MockTaskMetadata: MockTaskMetadata.return_value = Mock() service.add_task_metadata("task-1", "benchmark") MockTaskMetadata.assert_called_once_with( task_id="task-1", status="queued", task_type="benchmark", priority=0, ) def test_creates_queue_status_if_none_exists(self): """When no QueueStatus exists, _increment_queue_count creates one.""" from local_deep_research.database.queue_service import UserQueueService mock_session = Mock() mock_session.query.return_value.first.return_value = None service = UserQueueService(mock_session) with patch( "local_deep_research.database.queue_service.TaskMetadata" ) as MockTaskMetadata: MockTaskMetadata.return_value = Mock() with patch( "local_deep_research.database.queue_service.QueueStatus" ) as MockQueueStatus: mock_new_status = Mock() mock_new_status.queued_tasks = 0 MockQueueStatus.return_value = mock_new_status service.add_task_metadata("task-new", "research") # QueueStatus should have been created MockQueueStatus.assert_called_once_with( queued_tasks=0, active_tasks=0 ) # And queued_tasks incremented from 0 to 1 assert mock_new_status.queued_tasks == 1 class TestUpdateTaskStatusExtended: """Extended tests for update_task_status transitions.""" def _make_service_with_task(self, task_mock, status_mock): """Helper to set up a service with a findable task and status.""" mock_session = Mock() mock_filter = Mock() mock_filter.first.return_value = task_mock mock_task_query = Mock() mock_task_query.filter_by.return_value = mock_filter def query_side_effect(model): if hasattr(model, "task_id"): # TaskMetadata return mock_task_query # QueueStatus query for _get_or_create_status return Mock(first=Mock(return_value=status_mock)) mock_session.query.side_effect = query_side_effect return mock_session def test_queued_to_processing_sets_started_at_and_adjusts_counts(self): """Transition queued->processing should set started_at and adjust counts.""" from local_deep_research.database.queue_service import UserQueueService mock_task = Mock() mock_task.status = "queued" mock_task.started_at = None mock_status = Mock() mock_status.queued_tasks = 5 mock_status.active_tasks = 2 mock_session = self._make_service_with_task(mock_task, mock_status) service = UserQueueService(mock_session) service.update_task_status("task-1", "processing") assert mock_task.status == "processing" assert mock_task.started_at is not None # queued: max(0, 5 + (-1)) = 4, active: max(0, 2 + 1) = 3 assert mock_status.queued_tasks == 4 assert mock_status.active_tasks == 3 mock_session.commit.assert_called_once() def test_processing_to_completed_sets_completed_at_and_adjusts_counts(self): """Transition processing->completed should set completed_at and decrement active.""" from local_deep_research.database.queue_service import UserQueueService mock_task = Mock() mock_task.status = "processing" mock_task.completed_at = None mock_status = Mock() mock_status.queued_tasks = 3 mock_status.active_tasks = 2 mock_session = self._make_service_with_task(mock_task, mock_status) service = UserQueueService(mock_session) service.update_task_status("task-1", "completed") assert mock_task.status == "completed" assert mock_task.completed_at is not None assert mock_task.error_message is None # queued: max(0, 3 + 0) = 3, active: max(0, 2 + (-1)) = 1 assert mock_status.queued_tasks == 3 assert mock_status.active_tasks == 1 mock_session.commit.assert_called_once() def test_processing_to_failed_sets_completed_at_and_error_message(self): """Transition processing->failed should set completed_at, error_message, and decrement active.""" from local_deep_research.database.queue_service import UserQueueService mock_task = Mock() mock_task.status = "processing" mock_task.completed_at = None mock_task.error_message = None mock_status = Mock() mock_status.queued_tasks = 0 mock_status.active_tasks = 1 mock_session = self._make_service_with_task(mock_task, mock_status) service = UserQueueService(mock_session) service.update_task_status("task-1", "failed", "Timeout exceeded") assert mock_task.status == "failed" assert mock_task.error_message == "Timeout exceeded" assert mock_task.completed_at is not None # active: max(0, 1 + (-1)) = 0 assert mock_status.active_tasks == 0 mock_session.commit.assert_called_once() def test_task_not_found_does_nothing(self): """When task_id does not match any task, should do nothing.""" from local_deep_research.database.queue_service import UserQueueService mock_session = Mock() mock_filter = Mock() mock_filter.first.return_value = None mock_query = Mock() mock_query.filter_by.return_value = mock_filter mock_session.query.return_value = mock_query service = UserQueueService(mock_session) service.update_task_status("nonexistent-task", "completed") # Should not call commit since task was not found mock_session.commit.assert_not_called() def test_processing_to_completed_with_no_error_message(self): """Completing a task without error should set error_message to None.""" from local_deep_research.database.queue_service import UserQueueService mock_task = Mock() mock_task.status = "processing" mock_task.error_message = "old error" mock_status = Mock() mock_status.queued_tasks = 0 mock_status.active_tasks = 1 mock_session = self._make_service_with_task(mock_task, mock_status) service = UserQueueService(mock_session) service.update_task_status("task-1", "completed") # error_message should be set to None (the default) assert mock_task.error_message is None def test_non_transition_status_update(self): """Setting a status that doesn't match queued->processing or completed/failed should still update the status but not touch timestamps or counts.""" from local_deep_research.database.queue_service import UserQueueService mock_task = Mock() mock_task.status = "processing" mock_status = Mock() mock_status.queued_tasks = 2 mock_status.active_tasks = 3 mock_session = self._make_service_with_task(mock_task, mock_status) service = UserQueueService(mock_session) # A custom status that doesn't trigger any branch service.update_task_status("task-1", "cancelled") assert mock_task.status == "cancelled" # Counts should be unchanged since "cancelled" doesn't match any branch assert mock_status.queued_tasks == 2 assert mock_status.active_tasks == 3 class TestGetPendingTasksExtended: """Extended tests for get_pending_tasks method.""" def test_returns_ordered_list_of_task_dicts(self): """Should return list of dicts with task_id, task_type, created_at, priority.""" from local_deep_research.database.queue_service import UserQueueService mock_session = Mock() now = datetime.now(UTC) mock_task1 = Mock() mock_task1.task_id = "high-priority" mock_task1.task_type = "research" mock_task1.created_at = now mock_task1.priority = 10 mock_task2 = Mock() mock_task2.task_id = "low-priority" mock_task2.task_type = "benchmark" mock_task2.created_at = now mock_task2.priority = 1 mock_query = mock_session.query.return_value chain = mock_query.filter_by.return_value.order_by.return_value.limit.return_value chain.all.return_value = [mock_task1, mock_task2] service = UserQueueService(mock_session) result = service.get_pending_tasks(limit=10) assert len(result) == 2 assert result[0] == { "task_id": "high-priority", "task_type": "research", "created_at": now, "priority": 10, } assert result[1] == { "task_id": "low-priority", "task_type": "benchmark", "created_at": now, "priority": 1, } def test_returns_empty_list_when_no_pending_tasks(self): """Should return empty list when there are no queued tasks.""" from local_deep_research.database.queue_service import UserQueueService mock_session = Mock() mock_query = mock_session.query.return_value chain = mock_query.filter_by.return_value.order_by.return_value.limit.return_value chain.all.return_value = [] service = UserQueueService(mock_session) result = service.get_pending_tasks() assert result == [] def test_respects_limit_parameter(self): """Should pass the limit to the query chain.""" from local_deep_research.database.queue_service import UserQueueService mock_session = Mock() mock_query = mock_session.query.return_value mock_order = mock_query.filter_by.return_value.order_by.return_value mock_order.limit.return_value.all.return_value = [] service = UserQueueService(mock_session) service.get_pending_tasks(limit=5) mock_order.limit.assert_called_once_with(5) class TestCleanupOldTasksExtended: """Extended tests for cleanup_old_tasks method.""" def test_deletes_old_completed_and_failed_tasks(self): """Should delete tasks that are completed/failed and older than cutoff.""" from local_deep_research.database.queue_service import UserQueueService mock_session = Mock() mock_query = mock_session.query.return_value mock_query.filter.return_value.delete.return_value = 3 service = UserQueueService(mock_session) result = service.cleanup_old_tasks(days=14) assert result == 3 mock_session.commit.assert_called_once() def test_returns_zero_when_nothing_to_delete(self): """Should return 0 when no tasks match the criteria.""" from local_deep_research.database.queue_service import UserQueueService mock_session = Mock() mock_query = mock_session.query.return_value mock_query.filter.return_value.delete.return_value = 0 service = UserQueueService(mock_session) result = service.cleanup_old_tasks(days=7) assert result == 0 mock_session.commit.assert_called_once() def test_uses_correct_default_days(self): """Default days parameter should be 7 -- just verify it works without args.""" from local_deep_research.database.queue_service import UserQueueService mock_session = Mock() mock_query = mock_session.query.return_value mock_query.filter.return_value.delete.return_value = 0 service = UserQueueService(mock_session) # Call with no arguments to exercise the default days=7 result = service.cleanup_old_tasks() assert result == 0 mock_session.commit.assert_called_once() class TestUpdateQueueCountsExtended: """Extended tests for _update_queue_counts negative value protection.""" def test_prevents_negative_queued_tasks(self): """max(0, ...) should prevent queued_tasks from going negative.""" from local_deep_research.database.queue_service import UserQueueService mock_session = Mock() mock_status = Mock() mock_status.queued_tasks = 2 mock_status.active_tasks = 5 mock_session.query.return_value.first.return_value = mock_status service = UserQueueService(mock_session) service._update_queue_counts(-10, 0) assert mock_status.queued_tasks == 0 assert mock_status.active_tasks == 5 def test_prevents_negative_active_tasks(self): """max(0, ...) should prevent active_tasks from going negative.""" from local_deep_research.database.queue_service import UserQueueService mock_session = Mock() mock_status = Mock() mock_status.queued_tasks = 3 mock_status.active_tasks = 1 mock_session.query.return_value.first.return_value = mock_status service = UserQueueService(mock_session) service._update_queue_counts(0, -100) assert mock_status.queued_tasks == 3 assert mock_status.active_tasks == 0 def test_prevents_both_negative(self): """Both counts should be clamped to zero simultaneously.""" from local_deep_research.database.queue_service import UserQueueService mock_session = Mock() mock_status = Mock() mock_status.queued_tasks = 1 mock_status.active_tasks = 1 mock_session.query.return_value.first.return_value = mock_status service = UserQueueService(mock_session) service._update_queue_counts(-999, -999) assert mock_status.queued_tasks == 0 assert mock_status.active_tasks == 0 def test_positive_deltas_work_normally(self): """Positive deltas should increase counts normally.""" from local_deep_research.database.queue_service import UserQueueService mock_session = Mock() mock_status = Mock() mock_status.queued_tasks = 5 mock_status.active_tasks = 3 mock_session.query.return_value.first.return_value = mock_status service = UserQueueService(mock_session) service._update_queue_counts(2, 4) assert mock_status.queued_tasks == 7 assert mock_status.active_tasks == 7 def test_sets_last_checked_timestamp(self): """Should update last_checked when updating counts.""" from local_deep_research.database.queue_service import UserQueueService mock_session = Mock() mock_status = Mock() mock_status.queued_tasks = 0 mock_status.active_tasks = 0 mock_status.last_checked = None mock_session.query.return_value.first.return_value = mock_status service = UserQueueService(mock_session) service._update_queue_counts(1, 1) assert mock_status.last_checked is not None def test_creates_status_if_none_exists(self): """When no QueueStatus exists, should create one before updating.""" from local_deep_research.database.queue_service import UserQueueService mock_session = Mock() mock_session.query.return_value.first.return_value = None service = UserQueueService(mock_session) with patch( "local_deep_research.database.queue_service.QueueStatus" ) as MockQueueStatus: mock_new_status = Mock() mock_new_status.queued_tasks = 0 mock_new_status.active_tasks = 0 MockQueueStatus.return_value = mock_new_status service._update_queue_counts(3, 2) MockQueueStatus.assert_called_once_with( queued_tasks=0, active_tasks=0 ) mock_session.add.assert_called_once_with(mock_new_status) assert mock_new_status.queued_tasks == 3 assert mock_new_status.active_tasks == 2 class TestGetOrCreateStatus: """Tests for _get_or_create_status method.""" def test_returns_existing_status(self): """Should return existing QueueStatus when one exists.""" from local_deep_research.database.queue_service import UserQueueService mock_session = Mock() mock_status = Mock() mock_session.query.return_value.first.return_value = mock_status service = UserQueueService(mock_session) result = service._get_or_create_status() assert result is mock_status mock_session.add.assert_not_called() def test_creates_new_status_with_zero_counts(self): """When no status exists, should create one with zero counts.""" from local_deep_research.database.queue_service import UserQueueService mock_session = Mock() mock_session.query.return_value.first.return_value = None service = UserQueueService(mock_session) with patch( "local_deep_research.database.queue_service.QueueStatus" ) as MockQueueStatus: mock_new_status = Mock() MockQueueStatus.return_value = mock_new_status result = service._get_or_create_status() MockQueueStatus.assert_called_once_with( queued_tasks=0, active_tasks=0 ) mock_session.add.assert_called_once_with(mock_new_status) assert result is mock_new_status