Enhance Evidence Submission Error Handling and Logging

agent-framework/prometheus_swarm/utils/duplicate_evidence.py

@@ -0,0 +1,56 @@
+import logging
+from typing import Any, Dict, List
+
+class DuplicateEvidenceError(Exception):
+    """Custom exception for handling duplicate evidence scenarios."""
+    def __init__(self, message: str, existing_evidence: List[Dict[str, Any]] = None):
+        """
+        Initialize DuplicateEvidenceError with detailed information.
+
+        Args:
+            message (str): Description of the duplicate evidence error
+            existing_evidence (List[Dict[str, Any]], optional): List of existing evidence entries
+        """
+        self.message = message
+        self.existing_evidence = existing_evidence or []
+        super().__init__(self.message)
+
+def log_duplicate_evidence(evidence: Dict[str, Any], existing_entries: List[Dict[str, Any]]) -> None:
+    """
+    Log details about duplicate evidence.
+
+    Args:
+        evidence (Dict[str, Any]): The evidence being checked
+        existing_entries (List[Dict[str, Any]]): List of existing evidence entries
+    """
+    logger = logging.getLogger(__name__)
+    logger.warning(
+        "Duplicate evidence detected. "
+        f"Current Evidence: {evidence}, "
+        f"Existing Entries Count: {len(existing_entries)}, "
+        f"Existing Entries IDs: {[entry.get('id') for entry in existing_entries]}"
+    )
+
+def check_duplicate_evidence(new_evidence: Dict[str, Any], existing_evidence: List[Dict[str, Any]], unique_key: str = 'id') -> None:
+    """
+    Check for duplicate evidence and raise an error if duplicates are found.
+
+    Args:
+        new_evidence (Dict[str, Any]): The new evidence to be added
+        existing_evidence (List[Dict[str, Any]]): List of existing evidence
+        unique_key (str, optional): Key to use for identifying duplicates. Defaults to 'id'.
+
+    Raises:
+        DuplicateEvidenceError: If duplicate evidence is found
+    """
+    duplicates = [
+        entry for entry in existing_evidence 
+        if entry.get(unique_key) == new_evidence.get(unique_key)
+    ]
+
+    if duplicates:
+        log_duplicate_evidence(new_evidence, duplicates)
+        raise DuplicateEvidenceError(
+            f"Evidence with {unique_key} '{new_evidence.get(unique_key)}' already exists",
+            existing_evidence=duplicates
+        )

agent-framework/tests/unit/test_duplicate_evidence.py

@@ -0,0 +1,69 @@
+import pytest
+import logging
+from prometheus_swarm.utils.duplicate_evidence import (
+    DuplicateEvidenceError, 
+    check_duplicate_evidence, 
+    log_duplicate_evidence
+)
+
+def test_duplicate_evidence_error():
+    """Test the DuplicateEvidenceError is raised correctly."""
+    existing_evidence = [
+        {'id': 1, 'name': 'Evidence 1'},
+        {'id': 2, 'name': 'Evidence 2'}
+    ]
+    new_evidence = {'id': 1, 'name': 'Duplicate Evidence'}
+
+    with pytest.raises(DuplicateEvidenceError) as exc_info:
+        check_duplicate_evidence(new_evidence, existing_evidence)
+
+    assert str(exc_info.value) == "Evidence with id '1' already exists"
+    assert len(exc_info.value.existing_evidence) == 1
+
+def test_no_duplicate_evidence():
+    """Test that no error is raised when no duplicates exist."""
+    existing_evidence = [
+        {'id': 1, 'name': 'Evidence 1'},
+        {'id': 2, 'name': 'Evidence 2'}
+    ]
+    new_evidence = {'id': 3, 'name': 'New Evidence'}
+
+    try:
+        check_duplicate_evidence(new_evidence, existing_evidence)
+    except DuplicateEvidenceError:
+        pytest.fail("Unexpected DuplicateEvidenceError raised")
+
+def test_log_duplicate_evidence(caplog):
+    """Test logging of duplicate evidence."""
+    caplog.set_level(logging.WARNING)
+
+    existing_evidence = [
+        {'id': 1, 'name': 'Evidence 1'},
+        {'id': 2, 'name': 'Evidence 2'}
+    ]
+    new_evidence = {'id': 1, 'name': 'Duplicate Evidence'}
+
+    with pytest.raises(DuplicateEvidenceError):
+        check_duplicate_evidence(new_evidence, existing_evidence)
+
+    assert len(caplog.records) == 1
+    log_record = caplog.records[0]
+
+    assert log_record.levelname == 'WARNING'
+    assert 'Duplicate evidence detected' in log_record.msg
+    assert 'Current Evidence: ' + str(new_evidence) in log_record.msg
+    assert 'Existing Entries Count: 1' in log_record.msg
+    assert 'Existing Entries IDs: [1]' in log_record.msg
+
+def test_custom_unique_key():
+    """Test duplicate checking with a custom unique key."""
+    existing_evidence = [
+        {'key': 'abc', 'name': 'Evidence 1'},
+        {'key': 'def', 'name': 'Evidence 2'}
+    ]
+    new_evidence = {'key': 'abc', 'name': 'Duplicate Evidence'}
+
+    with pytest.raises(DuplicateEvidenceError) as exc_info:
+        check_duplicate_evidence(new_evidence, existing_evidence, unique_key='key')
+
+    assert str(exc_info.value) == "Evidence with key 'abc' already exists"