Implement Robust Error Handling for Duplicate Evidence Submissions

agent-framework/prometheus_swarm/utils/errors.py

@@ -1,16 +1,30 @@
-"""Error type for API errors."""
+import logging
+from typing import Any, Dict
 
+class DuplicateEvidenceError(Exception):
+    """
+    Custom exception raised when duplicate evidence is detected.
+
+    Attributes:
+        message (str): Description of the duplicate evidence error
+        evidence (Dict[str, Any]): The duplicate evidence details
+    """
+    def __init__(self, message: str, evidence: Dict[str, Any]):
+        self.message = message
+        self.evidence = evidence
+        super().__init__(self.message)
 
-class ClientAPIError(Exception):
-    """Error for API calls with status code."""
-
-    def __init__(self, original_error: Exception):
-        print(original_error)
-        if hasattr(original_error, "status_code"):
-            self.status_code = original_error.status_code
-        else:
-            self.status_code = 500
-        if hasattr(original_error, "message"):
-            super().__init__(original_error.message)
-        else:
-            super().__init__(str(original_error))
+def log_duplicate_evidence(evidence: Dict[str, Any]) -> None:
+    """
+    Log details of duplicate evidence with appropriate logging level.
+
+    Args:
+        evidence (Dict[str, Any]): Details of the duplicate evidence
+    """
+    logger = logging.getLogger('duplicate_evidence')
+    logger.warning(
+        f"Duplicate evidence detected: {evidence}",
+        extra={
+            'evidence': evidence
+        }
+    )

agent-framework/tests/unit/test_duplicate_evidence_error.py

@@ -0,0 +1,37 @@
+import logging
+import pytest
+from prometheus_swarm.utils.errors import DuplicateEvidenceError, log_duplicate_evidence
+
+def test_duplicate_evidence_error():
+    """Test the DuplicateEvidenceError exception."""
+    evidence = {
+        'id': '123',
+        'name': 'Test Evidence',
+        'type': 'document'
+    }
+
+    with pytest.raises(DuplicateEvidenceError) as exc_info:
+        raise DuplicateEvidenceError("Duplicate evidence found", evidence)
+
+    assert str(exc_info.value) == "Duplicate evidence found"
+    assert exc_info.value.evidence == evidence
+
+def test_log_duplicate_evidence(caplog):
+    """Test logging of duplicate evidence."""
+    caplog.set_level(logging.WARNING)
+
+    evidence = {
+        'id': '456',
+        'name': 'Another Test Evidence',
+        'type': 'record'
+    }
+
+    log_duplicate_evidence(evidence)
+
+    assert len(caplog.records) == 1
+    log_record = caplog.records[0]
+
+    assert log_record.levelno == logging.WARNING
+    assert 'Duplicate evidence detected' in log_record.getMessage()
+    assert evidence['id'] in log_record.getMessage()
+    assert evidence['name'] in log_record.getMessage()