""" Тесты для app.py и rag_system.py (без LLM и без ChromaDB). Запуск: python3 -m pytest test_app.py -v """ import os import sys import shutil import tempfile import unittest from pathlib import Path from unittest.mock import MagicMock, patch, PropertyMock # ── Добавляем директорию проекта в путь ────────────────────────────────────── sys.path.insert(0, str(Path(__file__).parent)) # ═══════════════════════════════════════════════════════════════════════ # Тесты: RAGAnswer (структура ответа) # ═══════════════════════════════════════════════════════════════════════ class TestRAGAnswer(unittest.TestCase): def _make(self): from rag_system import RAGAnswer return RAGAnswer def test_str_basic(self): """Базовый вывод: ответ + источники.""" RAGAnswer = self._make() a = RAGAnswer( answer="Бесплатно.", sources=[{"file": "doc.pdf", "page": 1, "score": 0.87, "snippet": "..."}], ) out = str(a) self.assertIn("Бесплатно.", out) self.assertIn("doc.pdf", out) self.assertIn("87%", out) def test_str_no_conflicts(self): """Без конфликтов — блок ПРОТИВОРЕЧИЯ не выводится.""" RAGAnswer = self._make() a = RAGAnswer(answer="OK", sources=[], conflicts=[]) self.assertNotIn("ПРОТИВОРЕЧИЯ", str(a)) def test_str_with_conflicts(self): """С конфликтами — предупреждение присутствует.""" RAGAnswer = self._make() a = RAGAnswer( answer="Разные данные.", sources=[], conflicts=["«4.5%» в fileA.pdf vs «4.7%» в fileB.pdf"], ) out = str(a) self.assertIn("ПРОТИВОРЕЧИЯ", out) self.assertIn("4.5%", out) self.assertIn("4.7%", out) def test_str_dedup_sources(self): """Одинаковые (файл, страница) показываются один раз.""" RAGAnswer = self._make() a = RAGAnswer( answer="X", sources=[ {"file": "a.pdf", "page": 1, "score": 0.9, "snippet": ""}, {"file": "a.pdf", "page": 1, "score": 0.8, "snippet": ""}, ], ) out = str(a) self.assertEqual(out.count("a.pdf"), 1) def test_score_none(self): """Если score не задан — не падает, выводит дефис.""" RAGAnswer = self._make() a = RAGAnswer( answer="X", sources=[{"file": "x.pdf", "page": 2, "snippet": ""}], # нет score ) out = str(a) self.assertIn("x.pdf", out) self.assertIn("—", out) # ═══════════════════════════════════════════════════════════════════════ # Тесты: _resolve_pdf_paths # ═══════════════════════════════════════════════════════════════════════ class TestResolvePdfPaths(unittest.TestCase): def _resolve(self, path): from rag_system import RAGSystem return RAGSystem._resolve_pdf_paths(path) def test_single_pdf_file(self): """Прямой путь к PDF возвращает список из одного элемента.""" with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as f: tmp = f.name try: result = self._resolve(tmp) self.assertEqual(result, [tmp]) finally: os.unlink(tmp) def test_non_pdf_file_ignored(self): """Файл не .pdf — возвращает пустой список.""" with tempfile.NamedTemporaryFile(suffix=".txt", delete=False) as f: tmp = f.name try: result = self._resolve(tmp) self.assertEqual(result, []) finally: os.unlink(tmp) def test_directory_with_pdfs(self): """Папка с PDF — находит все файлы рекурсивно.""" with tempfile.TemporaryDirectory() as d: p1 = Path(d) / "a.pdf" p2 = Path(d) / "sub" / "b.pdf" p2.parent.mkdir() p1.touch(); p2.touch() result = self._resolve(d) self.assertEqual(len(result), 2) self.assertTrue(all(r.endswith(".pdf") for r in result)) def test_empty_directory(self): """Пустая папка — пустой список.""" with tempfile.TemporaryDirectory() as d: self.assertEqual(self._resolve(d), []) def test_nonexistent_path(self): """Несуществующий путь — пустой список, не исключение.""" result = self._resolve("/nonexistent/path/to/nothing.pdf") self.assertEqual(result, []) # ═══════════════════════════════════════════════════════════════════════ # Тесты: _detect_conflicts # ═══════════════════════════════════════════════════════════════════════ class TestDetectConflicts(unittest.TestCase): def _make_doc(self, content: str, filename: str) -> object: from langchain_core.documents import Document return Document( page_content=content, metadata={"source_file": filename, "page": 0}, ) def _detect(self, docs_with_scores): """Вызывает _detect_conflicts без инициализации RAGSystem.""" from rag_system import RAGSystem fake = object.__new__(RAGSystem) return fake._detect_conflicts(docs_with_scores) def test_single_source_no_conflict(self): """Один источник — конфликтов быть не может.""" doc = self._make_doc("Комиссия 4.5% за операцию", "fileA.pdf") result = self._detect([(doc, 0.9)]) self.assertEqual(result, []) def test_same_value_no_conflict(self): """Два источника с одинаковым значением — не конфликт.""" d1 = self._make_doc("Ставка комиссия составляет 4.5%", "fileA.pdf") d2 = self._make_doc("Ставка комиссия составляет 4.5%", "fileB.pdf") result = self._detect([(d1, 0.9), (d2, 0.8)]) self.assertEqual(result, []) def test_different_percent_conflict(self): """Два источника с разными % при схожем контексте — конфликт.""" d1 = self._make_doc("комиссия платёж составляет 4.5%", "fileA.pdf") d2 = self._make_doc("комиссия платёж составляет 4.7%", "fileB.pdf") result = self._detect([(d1, 0.9), (d2, 0.8)]) self.assertTrue(len(result) >= 1) self.assertTrue(any("4.5" in r and "4.7" in r for r in result)) def test_different_units_no_conflict(self): """% и руб — разные единицы, конфликт не фиксируется.""" d1 = self._make_doc("комиссия сервис составляет 100 руб", "fileA.pdf") d2 = self._make_doc("комиссия сервис составляет 4.5%", "fileB.pdf") result = self._detect([(d1, 0.9), (d2, 0.8)]) self.assertEqual(result, []) def test_no_numeric_values_no_conflict(self): """Нет числовых значений — нет конфликтов.""" d1 = self._make_doc("Услуга предоставляется бесплатно.", "fileA.pdf") d2 = self._make_doc("Сервис недоступен в данный момент.", "fileB.pdf") result = self._detect([(d1, 0.9), (d2, 0.8)]) self.assertEqual(result, []) def test_deduplication(self): """Один и тот же конфликт не дублируется.""" content = "комиссия операция платёж 4.5%" d1a = self._make_doc(content, "fileA.pdf") d1b = self._make_doc(content, "fileA.pdf") d2 = self._make_doc("комиссия операция платёж 4.7%", "fileB.pdf") result = self._detect([(d1a, 0.9), (d1b, 0.85), (d2, 0.8)]) # Дубликаты удалены self.assertEqual(len(result), len(set(result))) # ═══════════════════════════════════════════════════════════════════════ # Тесты: similarity_search (порог релевантности) # ═══════════════════════════════════════════════════════════════════════ class TestSimilaritySearch(unittest.TestCase): def _make_rag_with_mock_store(self, scored_pairs): """Создаёт RAGSystem с замоканным vectorstore.""" from rag_system import RAGSystem from langchain_core.documents import Document rag = object.__new__(RAGSystem) mock_store = MagicMock() mock_store.similarity_search_with_relevance_scores.return_value = scored_pairs rag.vectorstore = mock_store rag._bm25_docs = [] return rag def test_all_above_threshold(self): """Все чанки выше порога — все возвращаются.""" from langchain_core.documents import Document doc = Document(page_content="text", metadata={}) rag = self._make_rag_with_mock_store([(doc, 0.9), (doc, 0.8), (doc, 0.5)]) result = rag.similarity_search("query", threshold=0.40) self.assertEqual(len(result), 3) def test_partial_filtering(self): """Часть чанков ниже порога — они отбрасываются.""" from langchain_core.documents import Document doc = Document(page_content="text", metadata={}) rag = self._make_rag_with_mock_store([ (doc, 0.9), (doc, 0.6), (doc, 0.2), (doc, 0.1) ]) result = rag.similarity_search("query", threshold=0.40) self.assertEqual(len(result), 2) self.assertTrue(all(s >= 0.40 for _, s in result)) def test_none_above_threshold_returns_best(self): """Если всё ниже порога — возвращается лучший чанк (fallback).""" from langchain_core.documents import Document doc = Document(page_content="text", metadata={}) rag = self._make_rag_with_mock_store([ (doc, 0.3), (doc, 0.2), (doc, 0.1) ]) result = rag.similarity_search("query", threshold=0.40) self.assertEqual(len(result), 1) self.assertEqual(result[0][1], 0.3) # лучший def test_empty_vectorstore(self): """Пустая база — возвращает пустой список.""" from rag_system import RAGSystem rag = object.__new__(RAGSystem) mock_store = MagicMock() mock_store.similarity_search_with_relevance_scores.return_value = [] rag.vectorstore = mock_store rag._bm25_docs = [] result = rag.similarity_search("query") self.assertEqual(result, []) def test_scores_preserved(self): """Scores в результате совпадают с исходными.""" from langchain_core.documents import Document doc = Document(page_content="x", metadata={}) pairs = [(doc, 0.95), (doc, 0.77), (doc, 0.55)] rag = self._make_rag_with_mock_store(pairs) result = rag.similarity_search("query", threshold=0.50) scores = [s for _, s in result] self.assertEqual(scores, [0.95, 0.77, 0.55]) # ═══════════════════════════════════════════════════════════════════════ # Тесты: app.py — функции без запуска сервера # ═══════════════════════════════════════════════════════════════════════ class TestAppFunctions(unittest.TestCase): def test_toggle_key_visibility_openai(self): """При выборе openai — поле ключа видимо.""" from app import toggle_key_visibility result = toggle_key_visibility("openai") self.assertTrue(result.get("visible", False)) def test_toggle_key_visibility_ollama(self): """При выборе ollama — поле ключа скрыто.""" from app import toggle_key_visibility result = toggle_key_visibility("ollama") self.assertFalse(result.get("visible", True)) def test_ask_empty_question(self): """Пустой вопрос возвращает подсказку, не вызывает RAG.""" from app import ask answer, sources = ask("", "ollama", "") self.assertIn("Введите", answer) self.assertEqual(sources, "") def test_ask_whitespace_question(self): """Вопрос из пробелов — тоже пустой.""" from app import ask answer, sources = ask(" \t\n ", "ollama", "") self.assertIn("Введите", answer) def test_index_pdf_no_files(self): """Индексация без файлов возвращает предупреждение.""" from app import index_pdf result = index_pdf(None, "ollama", "") self.assertIn("⚠️", result) def test_index_pdf_empty_list(self): """Пустой список файлов — предупреждение.""" from app import index_pdf result = index_pdf([], "ollama", "") self.assertIn("⚠️", result) def test_index_pdf_handles_rag_error(self): """Ошибка инициализации RAG (нет ключа) — красивое сообщение, не traceback.""" from app import index_pdf # Файл-заглушка fake_file = MagicMock() fake_file.name = "/nonexistent/file.pdf" result = index_pdf([fake_file], "openai", "") # openai без ключа self.assertTrue(result.startswith("❌")) def test_ask_rag_error_handled(self): """Ошибка внутри ask() — красивое сообщение, не падение.""" from app import ask # Сломаем _get_rag with patch("app._get_rag", side_effect=RuntimeError("тест ошибки")): answer, sources = ask("Вопрос", "ollama", "") self.assertTrue(answer.startswith("❌")) self.assertEqual(sources, "") def test_clear_db(self): """clear_db удаляет папку и сбрасывает _rag.""" import app from app import clear_db with tempfile.TemporaryDirectory() as d: # Патчим CHROMA_PERSIST_DIR чтобы не трогать реальную базу with patch("app.CHROMA_PERSIST_DIR", d): # Создаём тестовый файл в "базе" (Path(d) / "test.bin").touch() app._rag = MagicMock() # имитируем существующий RAG result = clear_db("ollama", "") self.assertIn("очищена", result.lower()) self.assertIsNone(app._rag) # Папка удалена self.assertFalse(Path(d).exists()) def test_clear_db_no_db_dir(self): """clear_db не падает если базы ещё нет.""" import app from app import clear_db with patch("app.CHROMA_PERSIST_DIR", "/nonexistent/chroma_db_test_xyz"): result = clear_db("ollama", "") self.assertIsNotNone(result) def test_get_rag_openai_no_key_raises(self): """_get_rag с openai и без ключа должен бросить ValueError.""" import app app._rag = None os.environ.pop("OPENAI_API_KEY", None) from app import _get_rag with self.assertRaises(Exception): _get_rag("openai", "") def test_get_rag_provider_switch_resets(self): """Смена провайдера пересоздаёт RAGSystem.""" import app mock_rag = MagicMock() mock_rag.llm_provider = "openai" app._rag = mock_rag with patch("app.RAGSystem") as MockRAG: MockRAG.return_value = MagicMock(llm_provider="ollama") from app import _get_rag result = _get_rag("ollama", "") MockRAG.assert_called_once_with(llm_provider="ollama") # ═══════════════════════════════════════════════════════════════════════ # Тесты: интеграция ask() с замоканным RAG # ═══════════════════════════════════════════════════════════════════════ class TestAskIntegration(unittest.TestCase): def _mock_rag_answer(self, answer_text, sources, conflicts=None): from rag_system import RAGAnswer return RAGAnswer( answer=answer_text, sources=sources, conflicts=conflicts or [], ) def test_ask_returns_answer_and_sources(self): """Нормальный ответ: answer и sources_md заполнены.""" from app import ask mock_answer = self._mock_rag_answer( "Бесплатно: электронный ключ.", [{"file": "sber.pdf", "page": 1, "score": 0.88, "snippet": ""}], ) mock_rag = MagicMock() mock_rag.ask_question.return_value = mock_answer with patch("app._get_rag", return_value=mock_rag): answer, sources = ask("Что бесплатно?", "ollama", "") self.assertIn("электронный ключ", answer) self.assertIn("sber.pdf", sources) self.assertIn("88%", sources) def test_ask_shows_conflicts_in_sources(self): """Конфликты отображаются в блоке источников.""" from app import ask mock_answer = self._mock_rag_answer( "Ставки различаются.", [{"file": "a.pdf", "page": 1, "score": 0.7, "snippet": ""}], conflicts=["«4.5%» в a.pdf vs «4.7%» в b.pdf"], ) mock_rag = MagicMock() mock_rag.ask_question.return_value = mock_answer with patch("app._get_rag", return_value=mock_rag): _, sources = ask("Какая ставка?", "ollama", "") self.assertIn("противоречия", sources.lower()) self.assertIn("4.5%", sources) def test_ask_no_sources_shows_placeholder(self): """Если источников нет — плейсхолдер вместо пустого блока.""" from app import ask mock_answer = self._mock_rag_answer("Информация не найдена.", []) mock_rag = MagicMock() mock_rag.ask_question.return_value = mock_answer with patch("app._get_rag", return_value=mock_rag): _, sources = ask("Что-то несуществующее", "ollama", "") self.assertIn("не найдены", sources.lower()) # ═══════════════════════════════════════════════════════════════════════ # Тесты: index_pdf с замоканным RAG # ═══════════════════════════════════════════════════════════════════════ class TestIndexPdf(unittest.TestCase): def test_index_pdf_success(self): """Успешная индексация — лог содержит имя файла и кол-во чанков.""" from app import index_pdf with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as f: tmp_path = f.name try: mock_rag = MagicMock() mock_rag.add_documents.return_value = 12 mock_rag.get_stats.return_value = {"total_chunks": 12} # Gradio 4+/5+/6+ передаёт файлы как объекты с .name fake_file = MagicMock() fake_file.name = tmp_path with patch("app._get_rag", return_value=mock_rag): result = index_pdf([fake_file], "ollama", "") self.assertIn("✅", result) self.assertIn("12", result) finally: os.unlink(tmp_path) def test_index_pdf_multiple_files(self): """Несколько файлов — каждый обрабатывается, итог в логе.""" from app import index_pdf files = [] paths = [] try: for i in range(3): f = tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) f.close() paths.append(f.name) fake = MagicMock() fake.name = f.name files.append(fake) mock_rag = MagicMock() mock_rag.add_documents.return_value = 5 mock_rag.get_stats.return_value = {"total_chunks": 15} with patch("app._get_rag", return_value=mock_rag): result = index_pdf(files, "ollama", "") self.assertEqual(result.count("✅"), 3) self.assertIn("15", result) finally: for p in paths: os.unlink(p) if __name__ == "__main__": unittest.main(verbosity=2)