#!/usr/bin/env python3
import asyncio
import sys
import os
# プロジェクトルートをパスに追加
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from src.main import search_events_by_location
async def test_location_search():
"""場所検索機能のテスト"""
print("=== 場所検索テスト ===\n")
try:
# 1. 沖縄のイベント検索テスト
test_prefecture = "沖縄"
print(f"'{test_prefecture}' のイベントを検索中...")
location_result = await search_events_by_location(prefecture=test_prefecture, count=5)
if "error" in location_result:
print(f"エラー: {location_result['error']}")
else:
print("場所検索成功!")
print("結果:")
print(f" - 取得件数: {location_result.get('results_returned', 0)}")
print(f" - 総件数: {location_result.get('results_available', 0)}")
events = location_result.get('events', [])
if events:
print(f"\n{test_prefecture} の最新イベント:")
for i, event in enumerate(events[:3], 1):
title = event.get('title', 'タイトル不明')
place = event.get('place', '場所不明')
started_at = event.get('started_at', '')
print(f" {i}. {title}")
print(f" 場所: {place}")
print(f" 開催日: {started_at}")
print()
# レート制限対応: 1秒待機
await asyncio.sleep(1.5)
# 2. キーワード検索テスト
print("キーワード検索テスト...")
keyword_result = await search_events_by_location(keyword="Python", count=3)
if "error" in keyword_result:
print(f"キーワード検索エラー: {keyword_result['error']}")
else:
print("キーワード検索成功!")
print(f"Python関連イベント数: {keyword_result.get('results_available', 0)}")
events = keyword_result.get('events', [])
if events:
print("\nPython関連イベント:")
for i, event in enumerate(events, 1):
title = event.get('title', 'タイトル不明')
print(f" {i}. {title}")
# レート制限対応: 1秒待機
await asyncio.sleep(1.5)
# 3. 存在しない都道府県のテスト
print("\n存在しない都道府県でのテスト...")
invalid_result = await search_events_by_location(prefecture="存在しない県")
if "error" in invalid_result:
print("適切にエラーハンドリング")
print(f" エラー内容: {invalid_result['error'][:100]}...")
else:
print("エラーハンドリング失敗")
except Exception as e:
print(f"場所検索テストエラー: {e}")
async def run_location_tests():
"""全ての場所関連テストを実行"""
print("CONNPASS MCP Server - 場所検索テスト\n")
print("="*50 + "\n")
await test_location_search()
print("\n場所検索テスト完了!")
if __name__ == "__main__":
try:
asyncio.run(run_location_tests())
except KeyboardInterrupt:
print("\nテスト中断")
except Exception as e:
print(f"テストエラー: {e}")
