Skip to content

Add test for TaskQueue handling duplicate task IDs #7

@m-marinucci

Description

@m-marinucci

As suggested in PR #1 by @sourcery-ai[bot]:

No test exists for TaskQueue.add_task with duplicate task IDs.

Task: Add a test to verify how TaskQueue handles adding a task with a duplicate ID.

Expected behavior to test:

  • Does it replace the existing task?
  • Does it reject the duplicate?
  • Does it create multiple entries?

Example test:

def test_task_queue_duplicate_id(self):
    """Test TaskQueue behavior with duplicate task IDs."""
    queue = TaskQueue()
    
    task1 = Task("duplicate-id", "compute", {"data": "first"}, priority=1)
    task2 = Task("duplicate-id", "compute", {"data": "second"}, priority=5)
    
    queue.add_task(task1)
    queue.add_task(task2)
    
    # Verify expected behavior (adjust based on implementation)
    assert len(queue.tasks) == 1  # Or 2, depending on design
    retrieved = queue.get_task("duplicate-id")
    assert retrieved.payload["data"] == "second"  # If it replaces

File: tests/test_master_service.py

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions