add tests for generation analysis

Author: saichandrapandraju

tests/detectors/llm_judge/test_llm_judge_detector.py

@@ -9,15 +9,17 @@
 from detectors.llm_judge.scheme import (
     ContentAnalysisHttpRequest,
     ContentAnalysisResponse,
+    GenerationAnalysisHttpRequest,
+    GenerationAnalysisResponse,
 )
 
 # Import vLLM Judge components for mocking
 from vllm_judge import EvaluationResult
 from vllm_judge.exceptions import MetricNotFoundError
 
 
-class TestLLMJudgeDetector:
-    """Test suite for LLMJudgeDetector."""
+class TestLLMJudgeDetectorContentAnalysis:
+    """Test suite for LLMJudgeDetector content analysis."""
 
     @pytest.fixture
     def mock_judge_result(self) -> EvaluationResult:
@@ -73,6 +75,16 @@ def test_detector_initialization_unreachable_url(self) -> None:
             with pytest.raises(Exception, match="Failed to detect model"):
                 LLMJudgeDetector()
 
+    def test_close_detector(self, detector_with_mock_judge: Tuple[LLMJudgeDetector, AsyncMock]) -> None:
+        """Test closing the detector properly closes the judge."""
+        detector: LLMJudgeDetector
+        mock_judge: AsyncMock
+        detector, mock_judge = detector_with_mock_judge
+        
+        asyncio.run(detector.close())
+        
+        mock_judge.close.assert_called_once()
+
     def test_evaluate_single_content_basic_metric(self, detector_with_mock_judge: Tuple[LLMJudgeDetector, AsyncMock]) -> None:
         """Test basic evaluation with just a metric."""
         detector: LLMJudgeDetector
@@ -182,7 +194,7 @@ def test_run_single_content(self, detector_with_mock_judge: Tuple[LLMJudgeDetect
             detector_params={"metric": "safety"}
         )
 
-        result = asyncio.run(detector.run(request))
+        result = asyncio.run(detector.analyze_content(request))
 
         assert len(result) == 1
         assert len(result[0]) == 1
@@ -216,7 +228,7 @@ def test_run_multiple_contents(self, detector_with_mock_judge: Tuple[LLMJudgeDet
             detector_params={"metric": "safety"}
         )
 
-        result = asyncio.run(detector.run(request))
+        result = asyncio.run(detector.analyze_content(request))
 
         assert len(result) == 3
         for i, analysis_list in enumerate(result):
@@ -252,7 +264,7 @@ def test_run_with_custom_evaluation_params(self, detector_with_mock_judge: Tuple
             detector_params=custom_evaluation_params
         )
 
-        result = asyncio.run(detector.run(request))
+        result = asyncio.run(detector.analyze_content(request))
 
         # Verify complex parameters were passed correctly
         expected_call_params = {
@@ -285,3 +297,292 @@ def test_close_detector(self, detector_with_mock_judge: Tuple[LLMJudgeDetector,
         asyncio.run(detector.close())
 
         mock_judge.close.assert_called_once()
+
+
+class TestLLMJudgeDetectorGenerationAnalysis:
+    """Test suite for LLMJudgeDetector generation analysis."""
+    
+    @pytest.fixture
+    def mock_judge_result(self) -> EvaluationResult:
+        """Mock EvaluationResult for generation testing."""
+        return EvaluationResult(
+            decision="HELPFUL",
+            reasoning="This generated response is helpful and addresses the user's question appropriately.",
+            score=0.85,
+            metadata={"model": "test-model"}
+        )
+    
+    @pytest.fixture
+    def detector_with_mock_judge(self, mock_judge_result) -> Tuple[LLMJudgeDetector, AsyncMock]:
+        """Create detector with mocked Judge."""
+        with patch.dict(os.environ, {"VLLM_BASE_URL": "http://test:8000"}):
+            with patch('vllm_judge.Judge.from_url') as mock_judge_class:
+                # Create mock judge instance
+                mock_judge_instance = AsyncMock()
+                mock_judge_instance.evaluate = AsyncMock(return_value=mock_judge_result)
+                mock_judge_instance.config.model = "test-model"
+                mock_judge_instance.config.base_url = "http://test:8000"
+                mock_judge_instance.close = AsyncMock()
+                
+                mock_judge_class.return_value = mock_judge_instance
+                
+                detector = LLMJudgeDetector()
+                return detector, mock_judge_instance
+
+    def test_evaluate_single_generation_basic_metric(self, detector_with_mock_judge: Tuple[LLMJudgeDetector, AsyncMock]) -> None:
+        """Test basic generation evaluation with just a metric."""
+        detector: LLMJudgeDetector
+        mock_judge: AsyncMock
+        detector, mock_judge = detector_with_mock_judge
+        
+        prompt = "What is artificial intelligence?"
+        generated_text = "Artificial intelligence (AI) refers to the simulation of human intelligence in machines."
+        params = {"metric": "helpfulness"}
+        
+        result = asyncio.run(detector.evaluate_single_generation(prompt, generated_text, params))
+        
+        # Verify judge.evaluate was called correctly
+        mock_judge.evaluate.assert_called_once_with(
+            input=prompt,
+            content=generated_text,
+            metric="helpfulness"
+        )
+        
+        # Verify response format
+        assert isinstance(result, GenerationAnalysisResponse)
+        assert result.detection == "HELPFUL"
+        assert result.score == 0.85
+        assert result.detection_type == "llm_judge" 
+        assert "reasoning" in result.metadata
+
+    def test_evaluate_single_generation_full_parameters(self, detector_with_mock_judge: Tuple[LLMJudgeDetector, AsyncMock]) -> None:
+        """Test generation evaluation with all vLLM Judge parameters."""
+        detector: LLMJudgeDetector
+        mock_judge: AsyncMock
+        detector, mock_judge = detector_with_mock_judge
+        
+        prompt = "Explain quantum computing in simple terms"
+        generated_text = "Quantum computing uses quantum bits (qubits) that can exist in multiple states simultaneously, allowing for parallel processing of information."
+        params = {
+            "criteria": "accuracy, clarity, and completeness",
+            "rubric": "Score based on technical accuracy and accessibility",
+            "scale": [1, 10],
+            "examples": [{"input": "test prompt", "output": "example response"}],
+            "system_prompt": "You are evaluating educational content",
+            "context": "This is for a general audience explanation of {topic}",
+            "template_vars": {"topic": "quantum computing"}
+        }
+        
+        asyncio.run(detector.evaluate_single_generation(prompt, generated_text, params))
+        
+        # Verify all parameters were passed through
+        expected_call = {
+            "input": prompt,
+            "content": generated_text,
+            **params
+        }
+        mock_judge.evaluate.assert_called_once_with(**expected_call)
+
+    def test_evaluate_single_generation_criteria_without_metric(self, detector_with_mock_judge: Tuple[LLMJudgeDetector, AsyncMock]) -> None:
+        """Test generation evaluation with criteria but no metric (should default scale)."""
+        detector: LLMJudgeDetector
+        mock_judge: AsyncMock
+        detector, mock_judge = detector_with_mock_judge
+        
+        prompt = "Write a short story"
+        generated_text = "Once upon a time, there was a brave knight who saved a village from a dragon."
+        params = {
+            "criteria": "creativity and engagement",
+            "rubric": "Custom rubric for story evaluation"
+        }
+        
+        asyncio.run(detector.evaluate_single_generation(prompt, generated_text, params))
+        
+        # Should add default scale when criteria provided without metric
+        expected_params = {
+            "input": prompt,
+            "content": generated_text,
+            "criteria": "creativity and engagement",
+            "rubric": "Custom rubric for story evaluation",
+            "scale": (0, 1)
+        }
+        mock_judge.evaluate.assert_called_once_with(**expected_params)
+
+    def test_evaluate_single_generation_no_params(self, detector_with_mock_judge: Tuple[LLMJudgeDetector, AsyncMock]) -> None:
+        """Test generation evaluation with no parameters (should default to safety)."""
+        detector: LLMJudgeDetector
+        mock_judge: AsyncMock
+        detector, mock_judge = detector_with_mock_judge
+        
+        prompt = "Tell me about AI"
+        generated_text = "AI is a field of computer science focused on creating intelligent machines."
+        params = {}
+        
+        asyncio.run(detector.evaluate_single_generation(prompt, generated_text, params))
+        
+        # Should default to safety metric
+        expected_params = {
+            "input": prompt,
+            "content": generated_text,
+            "metric": "safety"
+        }
+        mock_judge.evaluate.assert_called_once_with(**expected_params)
+
+    def test_evaluate_single_generation_invalid_metric(self, detector_with_mock_judge: Tuple[LLMJudgeDetector, AsyncMock]) -> None:
+        """Test generation evaluation with invalid metric raises error."""
+        detector: LLMJudgeDetector
+        detector, _ = detector_with_mock_judge
+        
+        prompt = "Test prompt"
+        generated_text = "Test generation"
+        params = {"metric": "invalid_metric"}
+        
+        with pytest.raises(MetricNotFoundError, match="Metric 'invalid_metric' not found"):
+            asyncio.run(detector.evaluate_single_generation(prompt, generated_text, params))
+
+    def test_analyze_generation_basic_request(self, detector_with_mock_judge: Tuple[LLMJudgeDetector, AsyncMock]) -> None:
+        """Test the analyze_generation method with basic request."""
+        detector: LLMJudgeDetector
+        mock_judge: AsyncMock
+        detector, mock_judge = detector_with_mock_judge
+        
+        request = GenerationAnalysisHttpRequest(
+            prompt="What is machine learning?",
+            generated_text="Machine learning is a subset of AI that enables computers to learn from data without explicit programming.",
+            detector_params={"metric": "accuracy"}
+        )
+        
+        result = asyncio.run(detector.analyze_generation(request))
+        
+        # Verify judge.evaluate was called correctly
+        mock_judge.evaluate.assert_called_once_with(
+            input="What is machine learning?",
+            content="Machine learning is a subset of AI that enables computers to learn from data without explicit programming.",
+            metric="accuracy"
+        )
+        
+        # Verify response format
+        assert isinstance(result, GenerationAnalysisResponse)
+        assert result.detection == "HELPFUL"
+        assert result.score == 0.85
+        assert result.detection_type == "llm_judge"
+        assert "reasoning" in result.metadata
+        assert result.metadata["reasoning"] is not None
+
+    def test_analyze_generation_complex_request(self, detector_with_mock_judge: Tuple[LLMJudgeDetector, AsyncMock]) -> None:
+        """Test the analyze_generation method with complex parameters."""
+        detector: LLMJudgeDetector
+        mock_judge: AsyncMock
+        detector, mock_judge = detector_with_mock_judge
+        
+        request = GenerationAnalysisHttpRequest(
+            prompt="Explain the benefits and risks of artificial intelligence",
+            generated_text="AI offers significant benefits like improved efficiency and automation, but also poses risks such as job displacement and potential bias in decision-making systems.",
+            detector_params={
+                "criteria": "balance, accuracy, and completeness",
+                "rubric": {
+                    1.0: "Excellent balance of benefits and risks with high accuracy",
+                    0.8: "Good coverage with minor gaps",
+                    0.6: "Adequate but missing some key points",
+                    0.4: "Poor coverage or significant inaccuracies",
+                    0.0: "Completely inadequate or misleading"
+                },
+                "scale": [0, 1],
+                "context": "This is for an educational discussion about AI ethics"
+            }
+        )
+        
+        result = asyncio.run(detector.analyze_generation(request))
+        
+        # Verify complex parameters were passed correctly
+        expected_call_params = {
+            "input": request.prompt,
+            "content": request.generated_text,
+            **request.detector_params
+        }
+        mock_judge.evaluate.assert_called_once_with(**expected_call_params)
+        
+        # Verify response
+        assert isinstance(result, GenerationAnalysisResponse)
+        assert result.detection_type == "llm_judge"
+
+    def test_analyze_generation_empty_params(self, detector_with_mock_judge: Tuple[LLMJudgeDetector, AsyncMock]) -> None:
+        """Test analyze_generation with empty detector params (should default to safety)."""
+        detector: LLMJudgeDetector
+        mock_judge: AsyncMock
+        detector, mock_judge = detector_with_mock_judge
+        
+        request = GenerationAnalysisHttpRequest(
+            prompt="Hello, how are you?",
+            generated_text="I'm doing well, thank you for asking! How can I assist you today?",
+            detector_params={}
+        )
+        
+        result = asyncio.run(detector.analyze_generation(request))
+        
+        # Should default to safety metric
+        expected_params = {
+            "input": request.prompt,
+            "content": request.generated_text,
+            "metric": "safety"
+        }
+        mock_judge.evaluate.assert_called_once_with(**expected_params)
+        
+        assert isinstance(result, GenerationAnalysisResponse)
+        assert result.detection_type == "llm_judge"
+
+    def test_generation_analysis_with_numeric_score(self, detector_with_mock_judge: Tuple[LLMJudgeDetector, AsyncMock]) -> None:
+        """Test generation analysis handles numeric scores correctly."""
+        detector: LLMJudgeDetector
+        mock_judge: AsyncMock
+        detector, mock_judge = detector_with_mock_judge
+        
+        # Mock a numeric decision result
+        numeric_result = EvaluationResult(
+            decision=8.5,
+            reasoning="High quality response with good accuracy",
+            score=8.5,
+            metadata={"model": "test-model"}
+        )
+        mock_judge.evaluate.return_value = numeric_result
+        
+        request = GenerationAnalysisHttpRequest(
+            prompt="Explain photosynthesis",
+            generated_text="Photosynthesis is the process by which plants convert light energy into chemical energy.",
+            detector_params={"metric": "accuracy", "scale": [0, 10]}
+        )
+        
+        result = asyncio.run(detector.analyze_generation(request))
+        
+        assert isinstance(result, GenerationAnalysisResponse)
+        assert result.detection == "8.5"
+        assert result.score == 8.5
+        assert result.detection_type == "llm_judge"
+
+    def test_generation_analysis_with_none_score(self, detector_with_mock_judge: Tuple[LLMJudgeDetector, AsyncMock]) -> None:
+        """Test generation analysis handles None scores correctly."""
+        detector: LLMJudgeDetector
+        mock_judge: AsyncMock
+        detector, mock_judge = detector_with_mock_judge
+        
+        # Mock a result with None score
+        none_score_result = EvaluationResult(
+            decision="GOOD",
+            reasoning="Good quality response",
+            score=None,
+            metadata={"model": "test-model"}
+        )
+        mock_judge.evaluate.return_value = none_score_result
+        
+        request = GenerationAnalysisHttpRequest(
+            prompt="Test prompt",
+            generated_text="Test generation",
+            detector_params={"metric": "helpfulness"}
+        )
+        
+        result = asyncio.run(detector.analyze_generation(request))
+        
+        assert isinstance(result, GenerationAnalysisResponse)
+        assert result.detection == "GOOD"
+        assert result.score == 0.0  # Should default to 0.0 when score is None
+        assert result.detection_type == "llm_judge"