""" Tests for search.py TDD: 테스트 먼저 작성, 구현은 이후 모든 외부 의존성(embedding_service, supabase)은 mock으로 처리. """ import os from unittest.mock import MagicMock, call, patch import pytest # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_vector(dim: int = 1536) -> list[float]: """dim 차원의 더미 벡터 반환.""" return [0.1] * dim def _make_rpc_response(rows: list[dict]) -> MagicMock: """Supabase RPC 반환값을 흉내 내는 mock 객체 생성.""" response = MagicMock() response.data = rows return response def _make_semantic_row( content: str = "테스트 내용", similarity: float = 0.95, source: str = "test_source", title: str = "테스트 제목", document_id: str = "doc-001", ) -> dict: """semantic/keyword_search 결과 행 생성.""" return { "content": content, "similarity": similarity, "source": source, "title": title, "document_id": document_id, } def _make_hybrid_row( content: str = "테스트 내용", similarity: float = 0.9, combined_score: float = 0.85, source: str = "test_source", title: str = "테스트 제목", document_id: str = "doc-001", ) -> dict: """hybrid_search 결과 행 생성.""" return { "content": content, "similarity": similarity, "combined_score": combined_score, "source": source, "title": title, "document_id": document_id, } # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture(autouse=True) def set_env_vars(monkeypatch): """모든 테스트에서 필요한 환경변수를 설정.""" monkeypatch.setenv("OPENAI_API_KEY", "test-openai-key") monkeypatch.setenv("INSURO_SUPABASE_URL", "https://test.supabase.co") monkeypatch.setenv("INSURO_SUPABASE_SERVICE_ROLE_KEY", "test-service-role-key") # --------------------------------------------------------------------------- # Test cases: semantic_search() # --------------------------------------------------------------------------- class TestSemanticSearch: """semantic_search() 테스트.""" def test_calls_get_embedding(self): """semantic_search() 호출 시 get_embedding()이 쿼리 텍스트로 호출된다.""" from search import semantic_search mock_vector = _make_vector() mock_rpc_response = _make_rpc_response([_make_semantic_row()]) mock_execute = MagicMock(return_value=mock_rpc_response) with ( patch("search.get_embedding", return_value=mock_vector) as mock_embed, patch("search._get_supabase_client") as mock_client_factory, ): mock_client = MagicMock() mock_client_factory.return_value = mock_client mock_client.rpc.return_value.execute = mock_execute semantic_search("보험 약관 검색") mock_embed.assert_called_once_with("보험 약관 검색") def test_calls_supabase_rpc(self): """semantic_search() 호출 시 Supabase RPC가 호출된다.""" from search import semantic_search mock_vector = _make_vector() mock_rpc_response = _make_rpc_response([_make_semantic_row()]) mock_execute = MagicMock(return_value=mock_rpc_response) with ( patch("search.get_embedding", return_value=mock_vector), patch("search._get_supabase_client") as mock_client_factory, ): mock_client = MagicMock() mock_client_factory.return_value = mock_client mock_client.rpc.return_value.execute = mock_execute semantic_search("보험 약관 검색") # RPC가 호출되어야 함 assert mock_client.rpc.called def test_returns_correct_format(self): """semantic_search() 결과가 올바른 형식을 가진다.""" from search import semantic_search mock_vector = _make_vector() row = _make_semantic_row( content="보험 내용", similarity=0.92, source="insuro", title="약관 제목", document_id="doc-abc", ) mock_rpc_response = _make_rpc_response([row]) mock_execute = MagicMock(return_value=mock_rpc_response) with ( patch("search.get_embedding", return_value=mock_vector), patch("search._get_supabase_client") as mock_client_factory, ): mock_client = MagicMock() mock_client_factory.return_value = mock_client mock_client.rpc.return_value.execute = mock_execute results = semantic_search("보험") assert isinstance(results, list) assert len(results) == 1 item = results[0] assert isinstance(item["content"], str) assert isinstance(item["similarity"], float) assert isinstance(item["source"], str) assert isinstance(item["title"], str) assert isinstance(item["document_id"], str) assert item["content"] == "보험 내용" assert item["similarity"] == 0.92 assert item["source"] == "insuro" assert item["title"] == "약관 제목" assert item["document_id"] == "doc-abc" def test_empty_result_returns_empty_list(self): """결과가 없을 때 빈 리스트를 반환한다.""" from search import semantic_search mock_vector = _make_vector() mock_rpc_response = _make_rpc_response([]) mock_execute = MagicMock(return_value=mock_rpc_response) with ( patch("search.get_embedding", return_value=mock_vector), patch("search._get_supabase_client") as mock_client_factory, ): mock_client = MagicMock() mock_client_factory.return_value = mock_client mock_client.rpc.return_value.execute = mock_execute results = semantic_search("존재하지 않는 내용") assert results == [] def test_limit_parameter_passed(self): """limit 파라미터가 RPC 호출에 전달된다.""" from search import semantic_search mock_vector = _make_vector() mock_rpc_response = _make_rpc_response([]) mock_execute = MagicMock(return_value=mock_rpc_response) with ( patch("search.get_embedding", return_value=mock_vector), patch("search._get_supabase_client") as mock_client_factory, ): mock_client = MagicMock() mock_client_factory.return_value = mock_client mock_client.rpc.return_value.execute = mock_execute semantic_search("테스트", limit=5) # RPC 호출 인자에 match_count=5 또는 limit=5가 포함되어야 함 rpc_call_args = mock_client.rpc.call_args params = rpc_call_args[0][1] if rpc_call_args[0] else rpc_call_args[1] assert params.get("match_count") == 5 or params.get("limit") == 5 def test_source_filter_parameter_passed(self): """source_filter 파라미터가 RPC 호출에 전달된다.""" from search import semantic_search mock_vector = _make_vector() mock_rpc_response = _make_rpc_response([]) mock_execute = MagicMock(return_value=mock_rpc_response) with ( patch("search.get_embedding", return_value=mock_vector), patch("search._get_supabase_client") as mock_client_factory, ): mock_client = MagicMock() mock_client_factory.return_value = mock_client mock_client.rpc.return_value.execute = mock_execute semantic_search("테스트", source_filter="insuro") rpc_call_args = mock_client.rpc.call_args params = rpc_call_args[0][1] if rpc_call_args[0] else rpc_call_args[1] assert params.get("source_filter") == "insuro" def test_results_sorted_by_similarity_desc(self): """결과가 similarity 내림차순으로 정렬된다.""" from search import semantic_search mock_vector = _make_vector() rows = [ _make_semantic_row(content="낮은 유사도", similarity=0.5), _make_semantic_row(content="높은 유사도", similarity=0.95), _make_semantic_row(content="중간 유사도", similarity=0.7), ] mock_rpc_response = _make_rpc_response(rows) mock_execute = MagicMock(return_value=mock_rpc_response) with ( patch("search.get_embedding", return_value=mock_vector), patch("search._get_supabase_client") as mock_client_factory, ): mock_client = MagicMock() mock_client_factory.return_value = mock_client mock_client.rpc.return_value.execute = mock_execute results = semantic_search("테스트") similarities = [r["similarity"] for r in results] assert similarities == sorted(similarities, reverse=True) # --------------------------------------------------------------------------- # Test cases: keyword_search() # --------------------------------------------------------------------------- class TestKeywordSearch: """keyword_search() 테스트.""" def test_calls_supabase_rpc(self): """keyword_search() 호출 시 Supabase RPC가 호출된다.""" from search import keyword_search mock_rpc_response = _make_rpc_response([_make_semantic_row()]) mock_execute = MagicMock(return_value=mock_rpc_response) with patch("search._get_supabase_client") as mock_client_factory: mock_client = MagicMock() mock_client_factory.return_value = mock_client mock_client.rpc.return_value.execute = mock_execute keyword_search("보험료") assert mock_client.rpc.called def test_does_not_call_get_embedding(self): """keyword_search()는 get_embedding()을 호출하지 않는다.""" from search import keyword_search mock_rpc_response = _make_rpc_response([]) mock_execute = MagicMock(return_value=mock_rpc_response) with patch("search.get_embedding") as mock_embed, patch("search._get_supabase_client") as mock_client_factory: mock_client = MagicMock() mock_client_factory.return_value = mock_client mock_client.rpc.return_value.execute = mock_execute keyword_search("보험료") mock_embed.assert_not_called() def test_returns_correct_format(self): """keyword_search() 결과가 올바른 형식을 가진다.""" from search import keyword_search row = _make_semantic_row( content="키워드 내용", similarity=0.8, source="insuro", title="키워드 제목", document_id="doc-kw-001", ) mock_rpc_response = _make_rpc_response([row]) mock_execute = MagicMock(return_value=mock_rpc_response) with patch("search._get_supabase_client") as mock_client_factory: mock_client = MagicMock() mock_client_factory.return_value = mock_client mock_client.rpc.return_value.execute = mock_execute results = keyword_search("보험료") assert isinstance(results, list) assert len(results) == 1 item = results[0] assert "content" in item assert "source" in item assert "title" in item assert "document_id" in item def test_empty_result_returns_empty_list(self): """결과가 없을 때 빈 리스트를 반환한다.""" from search import keyword_search mock_rpc_response = _make_rpc_response([]) mock_execute = MagicMock(return_value=mock_rpc_response) with patch("search._get_supabase_client") as mock_client_factory: mock_client = MagicMock() mock_client_factory.return_value = mock_client mock_client.rpc.return_value.execute = mock_execute results = keyword_search("존재하지 않는 키워드") assert results == [] def test_limit_parameter_passed(self): """limit 파라미터가 RPC 호출에 전달된다.""" from search import keyword_search mock_rpc_response = _make_rpc_response([]) mock_execute = MagicMock(return_value=mock_rpc_response) with patch("search._get_supabase_client") as mock_client_factory: mock_client = MagicMock() mock_client_factory.return_value = mock_client mock_client.rpc.return_value.execute = mock_execute keyword_search("테스트", limit=3) rpc_call_args = mock_client.rpc.call_args params = rpc_call_args[0][1] if rpc_call_args[0] else rpc_call_args[1] assert params.get("match_count") == 3 or params.get("limit") == 3 def test_source_filter_parameter_passed(self): """source_filter 파라미터가 RPC 호출에 전달된다.""" from search import keyword_search mock_rpc_response = _make_rpc_response([]) mock_execute = MagicMock(return_value=mock_rpc_response) with patch("search._get_supabase_client") as mock_client_factory: mock_client = MagicMock() mock_client_factory.return_value = mock_client mock_client.rpc.return_value.execute = mock_execute keyword_search("테스트", source_filter="insuro") rpc_call_args = mock_client.rpc.call_args params = rpc_call_args[0][1] if rpc_call_args[0] else rpc_call_args[1] assert params.get("source_filter") == "insuro" # --------------------------------------------------------------------------- # Test cases: hybrid_search() # --------------------------------------------------------------------------- class TestHybridSearch: """hybrid_search() 테스트.""" def test_calls_get_embedding(self): """hybrid_search() 호출 시 get_embedding()이 쿼리 텍스트로 호출된다.""" from search import hybrid_search mock_vector = _make_vector() mock_rpc_response = _make_rpc_response([_make_hybrid_row()]) mock_execute = MagicMock(return_value=mock_rpc_response) with ( patch("search.get_embedding", return_value=mock_vector) as mock_embed, patch("search._get_supabase_client") as mock_client_factory, ): mock_client = MagicMock() mock_client_factory.return_value = mock_client mock_client.rpc.return_value.execute = mock_execute hybrid_search("복합 검색 쿼리") mock_embed.assert_called_once_with("복합 검색 쿼리") def test_calls_hybrid_search_rpc(self): """hybrid_search() 호출 시 Supabase RPC('hybrid_search')가 호출된다.""" from search import hybrid_search mock_vector = _make_vector() mock_rpc_response = _make_rpc_response([_make_hybrid_row()]) mock_execute = MagicMock(return_value=mock_rpc_response) with ( patch("search.get_embedding", return_value=mock_vector), patch("search._get_supabase_client") as mock_client_factory, ): mock_client = MagicMock() mock_client_factory.return_value = mock_client mock_client.rpc.return_value.execute = mock_execute hybrid_search("복합 검색 쿼리") # rpc 함수명이 "hybrid_search"여야 함 rpc_call_args = mock_client.rpc.call_args rpc_name = rpc_call_args[0][0] if rpc_call_args[0] else rpc_call_args[1].get("fn") assert rpc_name == "hybrid_search" def test_passes_query_text_and_embedding(self): """hybrid_search() 호출 시 query_text와 query_embedding이 전달된다.""" from search import hybrid_search mock_vector = _make_vector() mock_rpc_response = _make_rpc_response([]) mock_execute = MagicMock(return_value=mock_rpc_response) with ( patch("search.get_embedding", return_value=mock_vector), patch("search._get_supabase_client") as mock_client_factory, ): mock_client = MagicMock() mock_client_factory.return_value = mock_client mock_client.rpc.return_value.execute = mock_execute hybrid_search("복합 검색") rpc_call_args = mock_client.rpc.call_args params = rpc_call_args[0][1] if rpc_call_args[0] else rpc_call_args[1] assert params["query_text"] == "복합 검색" assert params["query_embedding"] == mock_vector def test_default_weights(self): """기본 가중치가 semantic_weight=0.7, keyword_weight=0.3이다.""" from search import hybrid_search mock_vector = _make_vector() mock_rpc_response = _make_rpc_response([]) mock_execute = MagicMock(return_value=mock_rpc_response) with ( patch("search.get_embedding", return_value=mock_vector), patch("search._get_supabase_client") as mock_client_factory, ): mock_client = MagicMock() mock_client_factory.return_value = mock_client mock_client.rpc.return_value.execute = mock_execute hybrid_search("테스트") rpc_call_args = mock_client.rpc.call_args params = rpc_call_args[0][1] if rpc_call_args[0] else rpc_call_args[1] assert params["semantic_weight"] == 0.7 assert params["keyword_weight"] == 0.3 def test_custom_weights(self): """커스텀 가중치 (0.5, 0.5)가 RPC 호출에 올바르게 전달된다.""" from search import hybrid_search mock_vector = _make_vector() mock_rpc_response = _make_rpc_response([]) mock_execute = MagicMock(return_value=mock_rpc_response) with ( patch("search.get_embedding", return_value=mock_vector), patch("search._get_supabase_client") as mock_client_factory, ): mock_client = MagicMock() mock_client_factory.return_value = mock_client mock_client.rpc.return_value.execute = mock_execute hybrid_search("테스트", semantic_weight=0.5, keyword_weight=0.5) rpc_call_args = mock_client.rpc.call_args params = rpc_call_args[0][1] if rpc_call_args[0] else rpc_call_args[1] assert params["semantic_weight"] == 0.5 assert params["keyword_weight"] == 0.5 def test_source_filter_parameter_passed(self): """source_filter 파라미터가 RPC 호출에 전달된다.""" from search import hybrid_search mock_vector = _make_vector() mock_rpc_response = _make_rpc_response([]) mock_execute = MagicMock(return_value=mock_rpc_response) with ( patch("search.get_embedding", return_value=mock_vector), patch("search._get_supabase_client") as mock_client_factory, ): mock_client = MagicMock() mock_client_factory.return_value = mock_client mock_client.rpc.return_value.execute = mock_execute hybrid_search("테스트", source_filter="insuro") rpc_call_args = mock_client.rpc.call_args params = rpc_call_args[0][1] if rpc_call_args[0] else rpc_call_args[1] assert params["source_filter"] == "insuro" def test_limit_parameter_passed(self): """limit 파라미터가 match_count로 RPC 호출에 전달된다.""" from search import hybrid_search mock_vector = _make_vector() mock_rpc_response = _make_rpc_response([]) mock_execute = MagicMock(return_value=mock_rpc_response) with ( patch("search.get_embedding", return_value=mock_vector), patch("search._get_supabase_client") as mock_client_factory, ): mock_client = MagicMock() mock_client_factory.return_value = mock_client mock_client.rpc.return_value.execute = mock_execute hybrid_search("테스트", limit=20) rpc_call_args = mock_client.rpc.call_args params = rpc_call_args[0][1] if rpc_call_args[0] else rpc_call_args[1] assert params["match_count"] == 20 def test_empty_result_returns_empty_list(self): """결과가 없을 때 빈 리스트를 반환한다.""" from search import hybrid_search mock_vector = _make_vector() mock_rpc_response = _make_rpc_response([]) mock_execute = MagicMock(return_value=mock_rpc_response) with ( patch("search.get_embedding", return_value=mock_vector), patch("search._get_supabase_client") as mock_client_factory, ): mock_client = MagicMock() mock_client_factory.return_value = mock_client mock_client.rpc.return_value.execute = mock_execute results = hybrid_search("없는 내용") assert results == [] def test_results_sorted_by_combined_score_desc(self): """결과가 combined_score 내림차순으로 정렬된다.""" from search import hybrid_search mock_vector = _make_vector() rows = [ _make_hybrid_row(content="낮은 점수", combined_score=0.4), _make_hybrid_row(content="높은 점수", combined_score=0.9), _make_hybrid_row(content="중간 점수", combined_score=0.65), ] mock_rpc_response = _make_rpc_response(rows) mock_execute = MagicMock(return_value=mock_rpc_response) with ( patch("search.get_embedding", return_value=mock_vector), patch("search._get_supabase_client") as mock_client_factory, ): mock_client = MagicMock() mock_client_factory.return_value = mock_client mock_client.rpc.return_value.execute = mock_execute results = hybrid_search("테스트") combined_scores = [r["combined_score"] for r in results] assert combined_scores == sorted(combined_scores, reverse=True) def test_returns_correct_format(self): """hybrid_search() 결과가 올바른 형식을 가진다.""" from search import hybrid_search mock_vector = _make_vector() row = _make_hybrid_row( content="혼합 내용", similarity=0.88, combined_score=0.82, source="insuro", title="혼합 제목", document_id="doc-hyb-001", ) mock_rpc_response = _make_rpc_response([row]) mock_execute = MagicMock(return_value=mock_rpc_response) with ( patch("search.get_embedding", return_value=mock_vector), patch("search._get_supabase_client") as mock_client_factory, ): mock_client = MagicMock() mock_client_factory.return_value = mock_client mock_client.rpc.return_value.execute = mock_execute results = hybrid_search("테스트") assert isinstance(results, list) assert len(results) == 1 item = results[0] assert isinstance(item["content"], str) assert isinstance(item["similarity"], float) assert isinstance(item["combined_score"], float) assert isinstance(item["source"], str) assert isinstance(item["title"], str) assert isinstance(item["document_id"], str) assert item["content"] == "혼합 내용" assert item["combined_score"] == 0.82 # --------------------------------------------------------------------------- # Test cases: Korean keyword search (trigram/ILIKE fallback) # --------------------------------------------------------------------------- class TestKoreanKeywordSearch: """한국어 키워드 검색 테스트 (trigram + ILIKE fallback).""" def test_korean_keyword_basic(self): """한국어 키워드 '보험설계사'로 검색 시 결과가 반환된다.""" from search import keyword_search row = _make_semantic_row( content="보험설계사의 역할과 책임에 대한 안내", similarity=0.8, source="insuro", title="보험설계사 가이드", document_id="doc-kr-001", ) mock_rpc_response = _make_rpc_response([row]) mock_execute = MagicMock(return_value=mock_rpc_response) with patch("search._get_supabase_client") as mock_client_factory: mock_client = MagicMock() mock_client_factory.return_value = mock_client mock_client.rpc.return_value.execute = mock_execute results = keyword_search("보험설계사") assert len(results) == 1 assert "보험설계사" in results[0]["content"] def test_korean_keyword_geumsobeob(self): """'금소법' 키워드 검색 테스트.""" from search import keyword_search row = _make_semantic_row( content="금소법(금융소비자보호법)에 따른 설명의무", similarity=0.75, source="insuro", title="금소법 안내", document_id="doc-kr-002", ) mock_rpc_response = _make_rpc_response([row]) mock_execute = MagicMock(return_value=mock_rpc_response) with patch("search._get_supabase_client") as mock_client_factory: mock_client = MagicMock() mock_client_factory.return_value = mock_client mock_client.rpc.return_value.execute = mock_execute results = keyword_search("금소법") assert len(results) == 1 assert "금소법" in results[0]["content"] def test_korean_keyword_empty_rpc_triggers_ilike_fallback(self): """RPC 결과가 비어있으면 ILIKE fallback이 실행된다.""" from search import keyword_search # RPC 결과 비어있음 mock_rpc_response = _make_rpc_response([]) mock_execute = MagicMock(return_value=mock_rpc_response) # ILIKE fallback 결과 fallback_response = MagicMock() fallback_response.data = [{"id": "chunk-001", "document_id": "doc-fb-001", "content": "청약철회 관련 안내"}] with patch("search._get_supabase_client") as mock_client_factory: mock_client = MagicMock() mock_client_factory.return_value = mock_client mock_client.rpc.return_value.execute = mock_execute # ILIKE fallback chain mock_from = MagicMock() mock_client.from_.return_value = mock_from mock_from.select.return_value = mock_from mock_from.ilike.return_value = mock_from mock_from.limit.return_value = mock_from mock_from.execute.return_value = fallback_response results = keyword_search("청약철회") assert len(results) == 1 assert "청약철회" in results[0]["content"] assert results[0]["similarity"] == 0.5 # ILIKE 기본 점수 def test_korean_keyword_no_fallback_when_rpc_has_results(self): """RPC 결과가 있으면 ILIKE fallback이 실행되지 않는다.""" from search import keyword_search row = _make_semantic_row(content="보험료 납입 안내", similarity=0.9) mock_rpc_response = _make_rpc_response([row]) mock_execute = MagicMock(return_value=mock_rpc_response) with patch("search._get_supabase_client") as mock_client_factory: mock_client = MagicMock() mock_client_factory.return_value = mock_client mock_client.rpc.return_value.execute = mock_execute results = keyword_search("보험료") assert len(results) == 1 # from_ 호출이 없어야 함 (fallback 미실행) mock_client.from_.assert_not_called()