#!/usr/bin/env python3
"""レート制限の動作確認テスト"""
import asyncio
import sys
import os
from datetime import datetime
# プロジェクトルートをパスに追加
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from src.api_client import ConnpassClient
async def test_rate_limit():
"""
レート制限が正しく動作するかテストします。
3回連続でAPIリクエストを送り、全体の実行時間が適切かを確認。
"""
print("=== レート制限テスト ===\n")
print("3回連続でAPIリクエストを送信し、レート制限が機能するか確認します。")
print("3リクエストで最低2秒かかるはずです（リクエスト間隔が1秒）。\n")
client = ConnpassClient()
overall_start = datetime.now()
# 3回連続でリクエスト
for i in range(3):
print(f" リクエスト {i+1}/3 を送信中...")
start_time = datetime.now()
result = await client.get_events(keyword="Python", count=1)
elapsed = (datetime.now() - start_time).total_seconds()
if "error" in result:
print(f" エラー: {result['error']}")
else:
print(f" 成功 (処理時間: {elapsed:.2f}秒)")
overall_end = datetime.now()
total_time = (overall_end - overall_start).total_seconds()
# 検証: 3リクエストなので最低2秒かかるはず
expected_min_time = 2.0
all_valid = total_time >= expected_min_time
print(f"\n レート制限の検証:")
print(f" 全体の実行時間: {total_time:.2f}秒")
print(f" 期待される最小時間: {expected_min_time:.2f}秒")
print(f"\n{'' if all_valid else ''} レート制限テスト: {'合格' if all_valid else '不合格'}")
if all_valid:
print(" レート制限が正しく機能しています。")
else:
print(" レート制限が機能していない可能性があります。")
return all_valid
async def test_concurrent_requests():
"""
並行リクエストでもレート制限が機能するかテストします。
"""
print("\n" + "="*50 + "\n")
print("=== 並行リクエストテスト ===\n")
print("3つのリクエストを同時に開始し、レート制限により順次実行されることを確認します。\n")
client = ConnpassClient()
async def make_request(request_id: int):
"""個別のリクエストを実行"""
start_time = datetime.now()
print(f" リクエスト{request_id} 開始: {start_time.strftime('%H:%M:%S.%f')[:-3]}")
result = await client.get_events(keyword=f"Test{request_id}", count=1)
end_time = datetime.now()
elapsed = (end_time - start_time).total_seconds()
if "error" in result:
print(f" リクエスト{request_id} 完了: {end_time.strftime('%H:%M:%S.%f')[:-3]} (待機時間: {elapsed:.2f}秒)")
else:
print(f" リクエスト{request_id} 完了: {end_time.strftime('%H:%M:%S.%f')[:-3]} (待機時間: {elapsed:.2f}秒)")
return start_time, end_time
# 3つのリクエストを並行実行
overall_start = datetime.now()
results = await asyncio.gather(
make_request(1),
make_request(2),
make_request(3)
)
overall_end = datetime.now()
total_time = (overall_end - overall_start).total_seconds()
print(f"\n 結果:")
print(f" 全体の実行時間: {total_time:.2f}秒")
print(f" 期待される最小時間: 2.0秒 (3リクエスト × 1秒間隔)")
if total_time >= 2.0:
print(" レート制限が正しく機能しています")
return True
else:
print(" レート制限が機能していない可能性があります")
return False
async def main():
"""全テストを実行"""
print("Connpass API - レート制限テスト\n")
print("このテストは実際のAPIを使用します。")
print(" APIキーが設定されていることを確認してください。\n")
# 順次リクエストのテスト
test1_passed = await test_rate_limit()
await asyncio.sleep(2) # テスト間の待機
# 並行リクエストのテスト
test2_passed = await test_concurrent_requests()
print("\n" + "="*50)
print(f"\n{' すべてのテストに合格しました！' if test1_passed and test2_passed else ' 一部のテストが失敗しました'}\n")
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
print("\nテスト中断")
except Exception as e:
print(f" テストエラー: {e}")
import traceback
traceback.print_exc()
