From 4cc0763054fb17b5486a01c37060189ba5b8720f Mon Sep 17 00:00:00 2001 From: StatPan Date: Wed, 7 Jan 2026 23:46:07 +0900 Subject: [PATCH 1/3] test: reinforce QA and add comprehensive tool tests - Add test_qa_reinforcement.py for retry logic and feedback consistency - Add test_server_comprehensive.py for smart tools and combined reports - Add test_member_service.py for missing MemberService coverage - Refactor list_services to improve search flexibility (token-based) --- assemblymcp/services.py | 23 +++--- tests/test_member_service.py | 53 ++++++++++++++ tests/test_qa_reinforcement.py | 112 +++++++++++++++++++++++++++++ tests/test_server_comprehensive.py | 74 +++++++++++++++++++ uv.lock | 2 +- 5 files changed, 252 insertions(+), 12 deletions(-) create mode 100644 tests/test_member_service.py create mode 100644 tests/test_qa_reinforcement.py create mode 100644 tests/test_server_comprehensive.py diff --git a/assemblymcp/services.py b/assemblymcp/services.py index bf67bfe..4b50f65 100644 --- a/assemblymcp/services.py +++ b/assemblymcp/services.py @@ -116,7 +116,8 @@ async def list_services(self, keyword: str = "") -> list[dict[str, str]]: """ results = [] - search_keyword = re.sub(r"\s+", "", keyword).lower() if keyword else "" + # Split keyword into tokens for multi-word matching + search_tokens = keyword.lower().split() if keyword else [] # Iterate through all service metadata for service_id, metadata in self.client.service_metadata.items(): @@ -125,16 +126,16 @@ async def list_services(self, keyword: str = "") -> list[dict[str, str]]: category = metadata.get("category", "") # Filter by keyword if provided - if search_keyword: - normalized_name = re.sub(r"\s+", "", name).lower() - normalized_desc = re.sub(r"\s+", "", description).lower() - - if not ( - search_keyword in normalized_name - or search_keyword in normalized_desc - or search_keyword in service_id.lower() - ): - continue + if search_tokens: + target_text = f"{name} {description} {service_id}".lower() + + # All tokens must be present in the target text + if not all(token in target_text for token in search_tokens): + # Also try space-stripped matching as a fallback + stripped_target = re.sub(r"\s+", "", target_text) + stripped_keyword = re.sub(r"\s+", "", keyword).lower() + if stripped_keyword not in stripped_target: + continue results.append( { diff --git a/tests/test_member_service.py b/tests/test_member_service.py new file mode 100644 index 0000000..7728a1c --- /dev/null +++ b/tests/test_member_service.py @@ -0,0 +1,53 @@ +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from assemblymcp.services import MemberService + + +@pytest.mark.asyncio +async def test_member_info_filtering(): + """의원 성명 검색 시 공백 제거 및 필터링 로직 테스트""" + mock_client = MagicMock() + # 실제 API 구조 시뮬레이션: {"서비스ID": [{"head": [...]}, {"row": [...]}]} + mock_data = { + "OWSSC6001134T516707": [ + {"head": [{"list_total_count": 2}, {"RESULT": {"CODE": "INFO-000", "MESSAGE": "정상"}}]}, + {"row": [{"HG_NM": "홍 길 동", "POLY_NM": "A당"}, {"HG_NM": "김철수", "POLY_NM": "B당"}]}, + ] + } + + with patch("assemblymcp.services._get_data_with_retry", new_callable=AsyncMock) as mock_retry: + mock_retry.return_value = mock_data + service = MemberService(mock_client) + + # 1. 공백이 포함된 이름으로 검색 + results = await service.get_member_info("홍길동") + assert len(results) == 1 + assert "홍 길 동" in results[0]["HG_NM"] + + # 2. 검색 결과가 없는 경우 원본 반환 (기존 로직 유지 확인) + results = await service.get_member_info("이영희") + assert len(results) == 2 + + +@pytest.mark.asyncio +async def test_member_committee_careers_parsing(): + """의원 경력 데이터 파싱 테스트""" + mock_client = MagicMock() + # ORNDP7000993P115502 + mock_data = { + "ORNDP7000993P115502": [ + {"head": [{"list_total_count": 1}]}, + {"row": [{"HG_NM": "홍길동", "PROFILE_SJ": "법제사법위원회 위원", "FRTO_DATE": "2024.06.10 ~"}]}, + ] + } + + with patch("assemblymcp.services._get_data_with_retry", new_callable=AsyncMock) as mock_retry: + mock_retry.return_value = mock_data + service = MemberService(mock_client) + careers = await service.get_member_committee_careers("홍길동") + + assert len(careers) == 1 + assert careers[0].HG_NM == "홍길동" + assert "2024.06.10" in careers[0].FRTO_DATE diff --git a/tests/test_qa_reinforcement.py b/tests/test_qa_reinforcement.py new file mode 100644 index 0000000..f717d49 --- /dev/null +++ b/tests/test_qa_reinforcement.py @@ -0,0 +1,112 @@ +from unittest.mock import AsyncMock, MagicMock, patch + +import httpx +import pytest +from assembly_client.errors import AssemblyAPIError + +from assemblymcp.server import ( + get_committee_info, + get_member_info, + get_member_voting_history, + get_plenary_schedule, + search_bills, +) +from assemblymcp.services import DiscoveryService, normalize_age, normalize_unit_cd + + +# 1. Normalization & Utility Tests +def test_unit_cd_normalization(): + assert normalize_unit_cd("22") == "100022" + assert normalize_unit_cd(22) == "100022" + assert normalize_unit_cd("22대") == "100022" + assert normalize_unit_cd("100022") == "100022" + assert normalize_unit_cd(None) == "" + + +def test_age_normalization(): + assert normalize_age("100022") == "22" + assert normalize_age("22") == "22" + assert normalize_age(22) == "22" + assert normalize_age(None) == "" + + +@pytest.mark.asyncio +async def test_discovery_service_flexible_search(): + mock_client = MagicMock() + mock_client.service_metadata = { + "TEST_SERVICE": {"name": "위원회 위원 명단", "description": "설명", "category": "카테"} + } + service = DiscoveryService(mock_client) + + # 공백이 포함된 검색어로 공백 없는 서비스 찾기 + results = await service.list_services(keyword="위원회 명단") + assert len(results) == 1 + assert results[0]["id"] == "TEST_SERVICE" + + # 대소문자 및 부분 일치 + results = await service.list_services(keyword="test") + assert len(results) == 1 + + +# 2. Retry Mechanism Test +@pytest.mark.asyncio +async def test_api_retry_logic(): + from assemblymcp.services import _get_data_with_retry + + mock_client = MagicMock() + + # 2번 실패 후 3번째에 성공하는 시나리오 + side_effects = [AssemblyAPIError("First Fail"), httpx.RequestError("Second Fail"), {"SUCCESS": "DATA"}] + mock_client.get_data = AsyncMock(side_effect=side_effects) + + # 실제 호출 + result = await _get_data_with_retry(mock_client, "SERV_ID", {}) + + assert result == {"SUCCESS": "DATA"} + assert mock_client.get_data.call_count == 3 + + +# 3. Server Tool Feedback Consistency Tests +@pytest.mark.asyncio +async def test_tool_empty_result_messages(): + # search_bills + with patch("assemblymcp.server.bill_service") as mock_bill: + mock_bill.search_bills = AsyncMock(return_value=[]) + res = await search_bills.fn(keyword="no_data") + assert isinstance(res, str) + assert "검색 조건에 맞는 의안이 없습니다" in res + + # get_plenary_schedule + with patch("assemblymcp.server.meeting_service") as mock_meeting: + mock_meeting.get_plenary_schedule = AsyncMock(return_value=[]) + res = await get_plenary_schedule.fn(unit_cd="22") + assert isinstance(res, str) + assert "본회의 일정이 없습니다" in res + + # get_member_info + with patch("assemblymcp.server.member_service") as mock_member: + mock_member.get_member_info = AsyncMock(return_value=[]) + res = await get_member_info.fn(name="유령의원") + assert isinstance(res, str) + assert "정보를 찾을 수 없습니다" in res + + +# 4. Specific Validation & Error Logic +@pytest.mark.asyncio +async def test_voting_history_validation(): + # 파라미터 둘 다 없을 때 + res = await get_member_voting_history.fn(name=None, bill_id=None) + assert "반드시 입력해야 합니다" in res + + +@pytest.mark.asyncio +async def test_committee_info_error_handling(): + with patch("assemblymcp.server.committee_service") as mock_comm: + mock_comm.get_committee_list = AsyncMock(return_value=[]) + # 서비스에서 에러 딕셔너리 리턴 시나리오 + mock_comm.get_committee_members = AsyncMock( + return_value={"error": {"suggestion": "추천 검색어: 법제사법위원회"}} + ) + + res = await get_committee_info.fn(committee_name="잘못된위위") + assert "추천 검색어" in res diff --git a/tests/test_server_comprehensive.py b/tests/test_server_comprehensive.py new file mode 100644 index 0000000..1698bf0 --- /dev/null +++ b/tests/test_server_comprehensive.py @@ -0,0 +1,74 @@ +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from assemblymcp.server import ( + analyze_legislative_issue, + get_bill_voting_results, + get_legislative_reports, + get_member_info, + get_representative_report, +) + + +@pytest.mark.asyncio +async def test_analyze_legislative_issue_integration(): + """analyze_legislative_issue 도구가 각 서비스를 잘 조합하는지 테스트""" + with patch("assemblymcp.server.smart_service") as mock_smart: + mock_smart.analyze_legislative_issue = AsyncMock( + return_value={"topic": "인공지능", "summary": {"total_bills_found": 1}} + ) + + result = await analyze_legislative_issue.fn(topic="인공지능") + + assert result["topic"] == "인공지능" + mock_smart.analyze_legislative_issue.assert_called_once_with("인공지능", limit=5) + + +@pytest.mark.asyncio +async def test_get_representative_report_integration(): + """의원 종합 리포트 생성 도구 테스트""" + with patch("assemblymcp.server.smart_service") as mock_smart: + mock_report = MagicMock() + mock_report.model_dump.return_value = {"member_name": "홍길동", "stats": {}} + mock_smart.get_representative_report = AsyncMock(return_value=mock_report) + + result = await get_representative_report.fn(member_name="홍길동") + + assert result["member_name"] == "홍길동" + mock_smart.get_representative_report.assert_called_once_with("홍길동") + + +@pytest.mark.asyncio +async def test_get_member_info_flow(): + """의원 정보 검색 도구 흐름 테스트""" + with patch("assemblymcp.server.member_service") as mock_member: + mock_member.get_member_info = AsyncMock(return_value=[{"HG_NM": "홍길동"}]) + + result = await get_member_info.fn(name="홍길동") + + assert isinstance(result, list) + assert result[0]["HG_NM"] == "홍길동" + + +@pytest.mark.asyncio +async def test_get_legislative_reports_empty(): + """보고서 검색 결과가 없을 때 메시지 반환 테스트""" + with patch("assemblymcp.server.smart_service") as mock_smart: + mock_smart.get_legislative_reports = AsyncMock(return_value=[]) + + result = await get_legislative_reports.fn(keyword="우주전쟁") + + assert isinstance(result, str) + assert "보고서나 뉴스를 찾을 수 없습니다" in result + + +@pytest.mark.asyncio +async def test_get_bill_voting_results_not_found(): + """표결 결과가 없을 때 메시지 반환 테스트""" + with patch("assemblymcp.server.smart_service") as mock_smart: + mock_smart.get_bill_voting_results = AsyncMock(return_value={"message": "데이터 없음"}) + + result = await get_bill_voting_results.fn(bill_id="NON_EXIST") + + assert result == {"message": "데이터 없음"} # SmartService가 딕셔너리를 반환하는 경우 diff --git a/uv.lock b/uv.lock index 5da5926..e68ba4b 100644 --- a/uv.lock +++ b/uv.lock @@ -46,7 +46,7 @@ wheels = [ [[package]] name = "assemblymcp" -version = "0.4.0" +version = "0.4.2" source = { editable = "." } dependencies = [ { name = "assembly-api-client" }, From 054984060ce3300205634ec309a2215e5d34640a Mon Sep 17 00:00:00 2001 From: StatPan Date: Thu, 8 Jan 2026 00:00:01 +0900 Subject: [PATCH 2/3] refactor: optimize list_services search performance Move stripped_keyword calculation out of the metadata iteration loop to avoid redundant processing. --- assemblymcp/services.py | 43 ++++++++++++++++++++--------------------- 1 file changed, 21 insertions(+), 22 deletions(-) diff --git a/assemblymcp/services.py b/assemblymcp/services.py index 4b50f65..d2521c8 100644 --- a/assemblymcp/services.py +++ b/assemblymcp/services.py @@ -116,28 +116,27 @@ async def list_services(self, keyword: str = "") -> list[dict[str, str]]: """ results = [] - # Split keyword into tokens for multi-word matching - search_tokens = keyword.lower().split() if keyword else [] - - # Iterate through all service metadata - for service_id, metadata in self.client.service_metadata.items(): - name = metadata.get("name", "") - description = metadata.get("description", "") - category = metadata.get("category", "") - - # Filter by keyword if provided - if search_tokens: - target_text = f"{name} {description} {service_id}".lower() - - # All tokens must be present in the target text - if not all(token in target_text for token in search_tokens): - # Also try space-stripped matching as a fallback - stripped_target = re.sub(r"\s+", "", target_text) - stripped_keyword = re.sub(r"\s+", "", keyword).lower() - if stripped_keyword not in stripped_target: - continue - - results.append( + # Split keyword into tokens for multi-word matching + search_tokens = keyword.lower().split() if keyword else [] + stripped_keyword = re.sub(r"\s+", "", keyword).lower() if keyword else "" + + # Iterate through all service metadata + for service_id, metadata in self.client.service_metadata.items(): + name = metadata.get("name", "") + description = metadata.get("description", "") + category = metadata.get("category", "") + + # Filter by keyword if provided + if search_tokens: + target_text = f"{name} {description} {service_id}".lower() + + # All tokens must be present in the target text + if not all(token in target_text for token in search_tokens): + # Also try space-stripped matching as a fallback + stripped_target = re.sub(r"\s+", "", target_text) + if stripped_keyword not in stripped_target: + continue + results.append( { "id": service_id, "name": name, From f6660654535b6f1d2a9af16ada600bff1e8c184b Mon Sep 17 00:00:00 2001 From: StatPan Date: Thu, 8 Jan 2026 00:15:44 +0900 Subject: [PATCH 3/3] fix: resolve IndentationError and finalized comprehensive tests --- assemblymcp/services.py | 43 +++++++++++++++++++++-------------------- 1 file changed, 22 insertions(+), 21 deletions(-) diff --git a/assemblymcp/services.py b/assemblymcp/services.py index d2521c8..e78bd90 100644 --- a/assemblymcp/services.py +++ b/assemblymcp/services.py @@ -116,27 +116,28 @@ async def list_services(self, keyword: str = "") -> list[dict[str, str]]: """ results = [] - # Split keyword into tokens for multi-word matching - search_tokens = keyword.lower().split() if keyword else [] - stripped_keyword = re.sub(r"\s+", "", keyword).lower() if keyword else "" - - # Iterate through all service metadata - for service_id, metadata in self.client.service_metadata.items(): - name = metadata.get("name", "") - description = metadata.get("description", "") - category = metadata.get("category", "") - - # Filter by keyword if provided - if search_tokens: - target_text = f"{name} {description} {service_id}".lower() - - # All tokens must be present in the target text - if not all(token in target_text for token in search_tokens): - # Also try space-stripped matching as a fallback - stripped_target = re.sub(r"\s+", "", target_text) - if stripped_keyword not in stripped_target: - continue - results.append( + # Split keyword into tokens for multi-word matching + search_tokens = keyword.lower().split() if keyword else [] + stripped_keyword = re.sub(r"\s+", "", keyword).lower() if keyword else "" + + # Iterate through all service metadata + for service_id, metadata in self.client.service_metadata.items(): + name = metadata.get("name", "") + description = metadata.get("description", "") + category = metadata.get("category", "") + + # Filter by keyword if provided + if search_tokens: + target_text = f"{name} {description} {service_id}".lower() + + # All tokens must be present in the target text + if not all(token in target_text for token in search_tokens): + # Also try space-stripped matching as a fallback + stripped_target = re.sub(r"\s+", "", target_text) + if stripped_keyword not in stripped_target: + continue + + results.append( { "id": service_id, "name": name,