test: Add comprehensive coverage tests for document_vector_utils uncovered code paths

Author: Mermaid97

test/backend/test_document_vector_utils_coverage.py

@@ -26,7 +26,10 @@
     summarize_cluster_legacy,
     summarize_clusters_map_reduce,
     summarize_clusters,
-    merge_cluster_summaries
+    merge_cluster_summaries,
+    calculate_document_embedding,
+    auto_determine_k,
+    kmeans_cluster_documents
 )
 
 
@@ -308,3 +311,341 @@ def test_merge_cluster_summaries_single(self):
         assert isinstance(result, str)
         assert "Single cluster summary" in result
 
+
+class TestAdditionalCoverage:
+    """Test additional coverage for uncovered code paths"""
+    
+    def test_get_documents_from_es_non_list_documents(self):
+        """Test ES retrieval when all_documents is not a list"""
+        mock_es_core = MagicMock()
+        
+        # Mock the first search call to return a tuple instead of list
+        mock_es_core.client.search.side_effect = [
+            {
+                'aggregations': {
+                    'unique_documents': {
+                        'buckets': (  # This will trigger the isinstance check
+                            {'key': '/path/doc1.pdf', 'doc_count': 3},
+                        )
+                    }
+                }
+            },
+            {
+                'hits': {
+                    'hits': [
+                        {
+                            '_source': {
+                                'filename': 'doc1.pdf',
+                                'content': 'test content',
+                                'embedding': [0.1, 0.2, 0.3],
+                                'file_size': 1000
+                            }
+                        }
+                    ]
+                }
+            }
+        ]
+        
+        result = get_documents_from_es('test_index', mock_es_core)
+        assert isinstance(result, dict)
+    
+    def test_get_documents_from_es_no_chunks(self):
+        """Test ES retrieval when document has no chunks"""
+        mock_es_core = MagicMock()
+        mock_es_core.client.search.side_effect = [
+            {
+                'aggregations': {
+                    'unique_documents': {
+                        'buckets': [
+                            {'key': '/path/doc1.pdf', 'doc_count': 0}
+                        ]
+                    }
+                }
+            },
+            {
+                'hits': {
+                    'hits': []  # No chunks
+                }
+            }
+        ]
+        
+        result = get_documents_from_es('test_index', mock_es_core)
+        assert result == {}  # Should return empty dict when no chunks
+    
+    def test_calculate_document_embedding_exception(self):
+        """Test calculate_document_embedding with exception"""
+        chunks = [
+            {'content': 'test content', 'embedding': [0.1, 0.2, 0.3]}
+        ]
+        
+        # Mock numpy operations to raise exception
+        with patch('numpy.array') as mock_array:
+            mock_array.side_effect = Exception("Numpy error")
+            
+            result = calculate_document_embedding(chunks)
+            assert result is None
+    
+    def test_auto_determine_k_small_dataset(self):
+        """Test auto_determine_k with very small dataset"""
+        # Create embeddings with only 2 samples (less than min_k=3)
+        embeddings = np.array([[0.1, 0.2], [0.3, 0.4]])
+        
+        result = auto_determine_k(embeddings, min_k=3, max_k=5)
+        assert result == 2  # Should return max(2, n_samples)
+    
+    def test_auto_determine_k_exception(self):
+        """Test auto_determine_k with exception during calculation"""
+        embeddings = np.array([[0.1, 0.2], [0.3, 0.4], [0.5, 0.6]])
+        
+        # Mock silhouette_score to raise exception
+        with patch('sklearn.metrics.silhouette_score') as mock_silhouette:
+            mock_silhouette.side_effect = Exception("Silhouette error")
+            
+            result = auto_determine_k(embeddings, min_k=2, max_k=3)
+            # Should use heuristic fallback
+            assert isinstance(result, int)
+            assert result >= 2
+    
+    def test_kmeans_cluster_documents_empty(self):
+        """Test kmeans_cluster_documents with empty embeddings"""
+        result = kmeans_cluster_documents({})
+        assert result == {}
+    
+    def test_kmeans_cluster_documents_exception(self):
+        """Test kmeans_cluster_documents with exception"""
+        doc_embeddings = {
+            'doc1': np.array([0.1, 0.2, 0.3]),
+            'doc2': np.array([0.4, 0.5, 0.6])
+        }
+        
+        # Mock auto_determine_k to raise exception
+        with patch('backend.utils.document_vector_utils.auto_determine_k') as mock_auto_k:
+            mock_auto_k.side_effect = Exception("Auto K error")
+            
+            with pytest.raises(Exception, match="Failed to cluster documents"):
+                kmeans_cluster_documents(doc_embeddings)
+    
+    def test_process_documents_for_clustering_exception(self):
+        """Test process_documents_for_clustering with exception"""
+        mock_es_core = MagicMock()
+        mock_es_core.client.search.side_effect = Exception("ES error")
+        
+        with pytest.raises(Exception, match="Failed to process documents"):
+            process_documents_for_clustering('test_index', mock_es_core)
+    
+    def test_process_documents_for_clustering_no_embeddings(self):
+        """Test process_documents_for_clustering when some documents fail embedding calculation"""
+        mock_es_core = MagicMock()
+        mock_es_core.client.search.return_value = {
+            'aggregations': {
+                'unique_documents': {
+                    'buckets': [
+                        {'key': '/path/doc1.pdf', 'doc_count': 1}
+                    ]
+                }
+            },
+            'hits': {
+                'hits': [
+                    {
+                        '_source': {
+                            'filename': 'doc1.pdf',
+                            'content': 'test content',
+                            'embedding': [0.1, 0.2, 0.3],
+                            'file_size': 1000
+                        }
+                    }
+                ]
+            }
+        }
+        
+        # Mock calculate_document_embedding to return None
+        with patch('backend.utils.document_vector_utils.calculate_document_embedding') as mock_calc:
+            mock_calc.return_value = None
+            
+            docs, embeddings = process_documents_for_clustering('test_index', mock_es_core)
+            assert isinstance(docs, dict)
+            assert isinstance(embeddings, dict)
+            assert len(embeddings) == 0  # No successful embeddings
+    
+    def test_extract_cluster_content_missing_doc(self):
+        """Test extract_cluster_content with missing document"""
+        document_samples = {
+            'doc1': {
+                'chunks': [{'content': 'test content'}]
+            }
+        }
+        cluster_doc_ids = ['doc1', 'missing_doc']
+        
+        result = extract_cluster_content(document_samples, cluster_doc_ids)
+        assert isinstance(result, str)
+        assert 'test content' in result
+    
+    def test_extract_cluster_content_no_chunks(self):
+        """Test extract_cluster_content with document having no chunks"""
+        document_samples = {
+            'doc1': {
+                'chunks': []
+            }
+        }
+        cluster_doc_ids = ['doc1']
+        
+        result = extract_cluster_content(document_samples, cluster_doc_ids)
+        assert isinstance(result, str)
+    
+    def test_extract_representative_chunks_smart_import_error(self):
+        """Test extract_representative_chunks_smart with ImportError"""
+        chunks = [
+            {'content': 'chunk 1'},
+            {'content': 'chunk 2'},
+            {'content': 'chunk 3'}
+        ]
+        
+        # Mock the import to raise ImportError
+        with patch('builtins.__import__', side_effect=ImportError("Module not found")):
+            result = extract_representative_chunks_smart(chunks, max_chunks=2)
+            assert len(result) <= 2
+            assert len(result) > 0
+    
+    def test_extract_representative_chunks_smart_short_content(self):
+        """Test extract_representative_chunks_smart with short content"""
+        chunks = [
+            {'content': 'short'},
+            {'content': 'also short'},
+            {'content': 'very short content'}
+        ]
+        
+        result = extract_representative_chunks_smart(chunks, max_chunks=2)
+        assert len(result) <= 2
+        assert len(result) > 0
+    
+    def test_analyze_cluster_coherence_empty(self):
+        """Test analyze_cluster_coherence with empty cluster_doc_ids"""
+        document_samples = {
+            'doc1': {
+                'chunks': [{'content': 'test content'}]
+            }
+        }
+        cluster_doc_ids = []
+        
+        result = analyze_cluster_coherence(cluster_doc_ids, document_samples)
+        assert result == {}
+    
+    def test_analyze_cluster_coherence_missing_doc(self):
+        """Test analyze_cluster_coherence with missing document"""
+        document_samples = {
+            'doc1': {
+                'chunks': [{'content': 'test content'}]
+            }
+        }
+        cluster_doc_ids = ['doc1', 'missing_doc']
+        
+        result = analyze_cluster_coherence(cluster_doc_ids, document_samples)
+        assert isinstance(result, dict)
+    
+    def test_analyze_cluster_coherence_no_chunks(self):
+        """Test analyze_cluster_coherence with document having no chunks"""
+        document_samples = {
+            'doc1': {
+                'chunks': []
+            }
+        }
+        cluster_doc_ids = ['doc1']
+        
+        result = analyze_cluster_coherence(cluster_doc_ids, document_samples)
+        assert isinstance(result, dict)
+    
+    def test_summarize_clusters_map_reduce_missing_doc(self):
+        """Test summarize_clusters_map_reduce with missing document"""
+        document_samples = {
+            'doc1': {
+                'chunks': [{'content': 'test content'}],
+                'filename': 'test.pdf'
+            }
+        }
+        clusters = {0: ['doc1', 'missing_doc']}
+        
+        with patch('backend.utils.document_vector_utils.summarize_document') as mock_sum_doc:
+            mock_sum_doc.return_value = "Doc summary"
+            
+            with patch('backend.utils.document_vector_utils.summarize_cluster') as mock_sum_cluster:
+                mock_sum_cluster.return_value = "Cluster summary"
+                
+                result = summarize_clusters_map_reduce(document_samples, clusters)
+                assert isinstance(result, dict)
+                assert 0 in result
+    
+    def test_summarize_clusters_map_reduce_few_chunks(self):
+        """Test summarize_clusters_map_reduce with document having few chunks"""
+        document_samples = {
+            'doc1': {
+                'chunks': [
+                    {'content': 'chunk 1'},
+                    {'content': 'chunk 2'}
+                ],
+                'filename': 'test.pdf'
+            }
+        }
+        clusters = {0: ['doc1']}
+        
+        with patch('backend.utils.document_vector_utils.summarize_document') as mock_sum_doc:
+            mock_sum_doc.return_value = "Doc summary"
+            
+            with patch('backend.utils.document_vector_utils.summarize_cluster') as mock_sum_cluster:
+                mock_sum_cluster.return_value = "Cluster summary"
+                
+                result = summarize_clusters_map_reduce(document_samples, clusters)
+                assert isinstance(result, dict)
+                assert 0 in result
+    
+    def test_summarize_clusters_map_reduce_long_content(self):
+        """Test summarize_clusters_map_reduce with long content"""
+        long_content = 'x' * 1500  # Longer than 1000 chars
+        document_samples = {
+            'doc1': {
+                'chunks': [
+                    {'content': long_content}
+                ],
+                'filename': 'test.pdf'
+            }
+        }
+        clusters = {0: ['doc1']}
+        
+        with patch('backend.utils.document_vector_utils.summarize_document') as mock_sum_doc:
+            mock_sum_doc.return_value = "Doc summary"
+            
+            with patch('backend.utils.document_vector_utils.summarize_cluster') as mock_sum_cluster:
+                mock_sum_cluster.return_value = "Cluster summary"
+                
+                result = summarize_clusters_map_reduce(document_samples, clusters)
+                assert isinstance(result, dict)
+                assert 0 in result
+    
+    def test_summarize_clusters_map_reduce_no_valid_docs(self):
+        """Test summarize_clusters_map_reduce with no valid document summaries"""
+        document_samples = {
+            'doc1': {
+                'chunks': [{'content': 'test content'}],
+                'filename': 'test.pdf'
+            }
+        }
+        clusters = {0: ['doc1']}
+        
+        with patch('backend.utils.document_vector_utils.summarize_document') as mock_sum_doc:
+            mock_sum_doc.return_value = ""  # Empty summary
+            
+            with patch('backend.utils.document_vector_utils.summarize_cluster') as mock_sum_cluster:
+                mock_sum_cluster.return_value = "Cluster summary"
+                
+                result = summarize_clusters_map_reduce(document_samples, clusters)
+                assert isinstance(result, dict)
+                assert 0 in result
+    
+    def test_summarize_cluster_legacy_exception(self):
+        """Test summarize_cluster_legacy with exception"""
+        cluster_content = "Test cluster content"
+        
+        # Mock file operations to raise exception
+        with patch('builtins.open', side_effect=Exception("File error")):
+            result = summarize_cluster_legacy(cluster_content)
+            assert "Failed to generate summary" in result
+