| """ |
| Tests for text processor functionality. |
| """ |
|
|
| import unittest |
| from unittest.mock import patch, MagicMock |
| import sys |
| import os |
|
|
| |
| sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src')) |
|
|
| from src.utils.text_processor import TextProcessor |
|
|
| class TestTextProcessor(unittest.TestCase): |
| """Test cases for TextProcessor class.""" |
| |
| def setUp(self): |
| """Set up test fixtures.""" |
| self.api_key = "test_api_key" |
| self.processor = TextProcessor(self.api_key) |
| |
| def test_initialization(self): |
| """Test TextProcessor initialization.""" |
| self.assertEqual(self.processor.openai_api_key, self.api_key) |
| self.assertIsNotNone(self.processor.embeddings) |
| self.assertIsNotNone(self.processor.llm) |
| |
| def test_create_documents_from_text(self): |
| """Test document creation from text.""" |
| text = "This is a test transcript. It has multiple sentences." |
| metadata = {"video_id": "test123", "title": "Test Video"} |
| |
| documents = self.processor.create_documents_from_text(text, metadata) |
| |
| self.assertIsInstance(documents, list) |
| self.assertGreater(len(documents), 0) |
| |
| |
| first_doc = documents[0] |
| self.assertIn("test transcript", first_doc.page_content.lower()) |
| self.assertEqual(first_doc.metadata["video_id"], "test123") |
| self.assertEqual(first_doc.metadata["title"], "Test Video") |
| |
| def test_create_documents_from_text_no_metadata(self): |
| """Test document creation without metadata.""" |
| text = "Simple test text." |
| |
| documents = self.processor.create_documents_from_text(text) |
| |
| self.assertIsInstance(documents, list) |
| self.assertGreater(len(documents), 0) |
| self.assertEqual(documents[0].metadata, {}) |
| |
| @patch('src.utils.text_processor.FAISS') |
| def test_create_vector_store_success(self, mock_faiss): |
| """Test successful vector store creation.""" |
| mock_vectorstore = MagicMock() |
| mock_faiss.from_documents.return_value = mock_vectorstore |
| |
| documents = [MagicMock()] |
| result = self.processor.create_vector_store(documents) |
| |
| self.assertEqual(result, mock_vectorstore) |
| mock_faiss.from_documents.assert_called_once_with(documents, self.processor.embeddings) |
| |
| def test_create_vector_store_empty_documents(self): |
| """Test vector store creation with empty documents.""" |
| result = self.processor.create_vector_store([]) |
| self.assertIsNone(result) |
| |
| @patch('src.utils.text_processor.FAISS') |
| def test_create_vector_store_failure(self, mock_faiss): |
| """Test vector store creation failure.""" |
| mock_faiss.from_documents.side_effect = Exception("Test error") |
| |
| documents = [MagicMock()] |
| result = self.processor.create_vector_store(documents) |
| |
| self.assertIsNone(result) |
| |
| @patch('src.utils.text_processor.RetrievalQA') |
| def test_create_qa_chain_success(self, mock_retrieval_qa): |
| """Test successful QA chain creation.""" |
| mock_qa_chain = MagicMock() |
| mock_retrieval_qa.from_chain_type.return_value = mock_qa_chain |
| |
| mock_vectorstore = MagicMock() |
| mock_retriever = MagicMock() |
| mock_vectorstore.as_retriever.return_value = mock_retriever |
| |
| result = self.processor.create_qa_chain(mock_vectorstore) |
| |
| self.assertEqual(result, mock_qa_chain) |
| mock_vectorstore.as_retriever.assert_called_once() |
| mock_retrieval_qa.from_chain_type.assert_called_once() |
| |
| @patch('src.utils.text_processor.RetrievalQA') |
| def test_create_qa_chain_failure(self, mock_retrieval_qa): |
| """Test QA chain creation failure.""" |
| mock_retrieval_qa.from_chain_type.side_effect = Exception("Test error") |
| |
| mock_vectorstore = MagicMock() |
| result = self.processor.create_qa_chain(mock_vectorstore) |
| |
| self.assertIsNone(result) |
| |
| def test_ask_question_success(self): |
| """Test successful question asking.""" |
| mock_qa_chain = MagicMock() |
| mock_qa_chain.return_value = { |
| 'result': 'Test answer', |
| 'source_documents': [MagicMock()] |
| } |
| |
| question = "What is this about?" |
| result = self.processor.ask_question(mock_qa_chain, question) |
| |
| self.assertTrue(result['success']) |
| self.assertEqual(result['answer'], 'Test answer') |
| self.assertIsNotNone(result['source_documents']) |
| self.assertIsNone(result['error']) |
| |
| mock_qa_chain.assert_called_once_with({"query": question}) |
| |
| def test_ask_question_failure(self): |
| """Test question asking failure.""" |
| mock_qa_chain = MagicMock() |
| mock_qa_chain.side_effect = Exception("Test error") |
| |
| question = "What is this about?" |
| result = self.processor.ask_question(mock_qa_chain, question) |
| |
| self.assertFalse(result['success']) |
| self.assertIsNone(result['answer']) |
| self.assertEqual(result['source_documents'], []) |
| self.assertIsNotNone(result['error']) |
| |
| @patch.object(TextProcessor, 'create_qa_chain') |
| @patch.object(TextProcessor, 'create_vector_store') |
| @patch.object(TextProcessor, 'create_documents_from_text') |
| def test_process_transcript_success(self, mock_create_docs, mock_create_vs, mock_create_qa): |
| """Test successful transcript processing.""" |
| |
| mock_documents = [MagicMock()] |
| mock_vectorstore = MagicMock() |
| mock_qa_chain = MagicMock() |
| |
| mock_create_docs.return_value = mock_documents |
| mock_create_vs.return_value = mock_vectorstore |
| mock_create_qa.return_value = mock_qa_chain |
| |
| transcript_text = "Test transcript text" |
| metadata = {"video_id": "test123"} |
| |
| result = self.processor.process_transcript(transcript_text, metadata) |
| |
| self.assertTrue(result['success']) |
| self.assertEqual(result['qa_chain'], mock_qa_chain) |
| self.assertEqual(result['vectorstore'], mock_vectorstore) |
| self.assertEqual(result['documents'], mock_documents) |
| self.assertIsNone(result['error']) |
| |
| mock_create_docs.assert_called_once_with(transcript_text, metadata) |
| mock_create_vs.assert_called_once_with(mock_documents) |
| mock_create_qa.assert_called_once_with(mock_vectorstore) |
| |
| @patch.object(TextProcessor, 'create_documents_from_text') |
| def test_process_transcript_document_creation_failure(self, mock_create_docs): |
| """Test transcript processing with document creation failure.""" |
| mock_create_docs.return_value = [] |
| |
| transcript_text = "Test transcript text" |
| result = self.processor.process_transcript(transcript_text) |
| |
| self.assertFalse(result['success']) |
| self.assertIsNone(result['qa_chain']) |
| self.assertIsNone(result['vectorstore']) |
| self.assertIsNone(result['documents']) |
| self.assertEqual(result['error'], "Failed to create documents from transcript") |
| |
| @patch.object(TextProcessor, 'create_vector_store') |
| @patch.object(TextProcessor, 'create_documents_from_text') |
| def test_process_transcript_vectorstore_creation_failure(self, mock_create_docs, mock_create_vs): |
| """Test transcript processing with vector store creation failure.""" |
| mock_create_docs.return_value = [MagicMock()] |
| mock_create_vs.return_value = None |
| |
| transcript_text = "Test transcript text" |
| result = self.processor.process_transcript(transcript_text) |
| |
| self.assertFalse(result['success']) |
| self.assertIsNone(result['qa_chain']) |
| self.assertIsNone(result['vectorstore']) |
| self.assertIsNotNone(result['documents']) |
| self.assertEqual(result['error'], "Failed to create vector store") |
| |
| @patch.object(TextProcessor, 'create_qa_chain') |
| @patch.object(TextProcessor, 'create_vector_store') |
| @patch.object(TextProcessor, 'create_documents_from_text') |
| def test_process_transcript_qa_chain_creation_failure(self, mock_create_docs, mock_create_vs, mock_create_qa): |
| """Test transcript processing with QA chain creation failure.""" |
| mock_create_docs.return_value = [MagicMock()] |
| mock_create_vs.return_value = MagicMock() |
| mock_create_qa.return_value = None |
| |
| transcript_text = "Test transcript text" |
| result = self.processor.process_transcript(transcript_text) |
| |
| self.assertFalse(result['success']) |
| self.assertIsNone(result['qa_chain']) |
| self.assertIsNotNone(result['vectorstore']) |
| self.assertIsNotNone(result['documents']) |
| self.assertEqual(result['error'], "Failed to create QA chain") |
|
|
| if __name__ == '__main__': |
| unittest.main() |
|
|
