File size: 4,204 Bytes
ac649ea |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 |
"""
test_news_tools.py
Test for news and official source tools (no social media sessions required)
"""
import sys
import os
import io
import json
# Force UTF-8 output
if sys.platform == 'win32':
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
print("="*60)
print("[TEST] NEWS & OFFICIAL SOURCE TOOLS")
print("="*60)
try:
from src.utils.tool_factory import create_tool_set
print("[OK] Tool factory imported")
tools = create_tool_set(include_profile_scrapers=False)
print("[OK] ToolSet created")
except Exception as e:
print(f"[FAIL] Could not create ToolSet: {e}")
sys.exit(1)
# Test 1: Local News
print("\n[1] Testing scrape_local_news...")
try:
local_news = tools.get("scrape_local_news")
if local_news:
result = local_news.invoke({"keywords": ["sri lanka"], "max_articles": 3})
parsed = json.loads(result) if isinstance(result, str) else result
if "error" in parsed:
print(f"[WARN] Local News returned error: {parsed['error']}")
else:
count = len(parsed.get('results', []))
print(f"[OK] Local News: {count} articles fetched")
if count > 0:
print(f" Sample: {str(parsed['results'][0])[:150]}...")
else:
print("[WARN] scrape_local_news not found in ToolSet")
except Exception as e:
print(f"[FAIL] Local News: {e}")
# Test 2: CSE Stock Data
print("\n[2] Testing scrape_cse_stock_data...")
try:
cse_tool = tools.get("scrape_cse_stock_data")
if cse_tool:
result = cse_tool.invoke({"symbol": "ASPI", "period": "1d"})
parsed = json.loads(result) if isinstance(result, str) else result
if "error" in parsed:
print(f"[WARN] CSE Stock returned error: {parsed['error']}")
else:
print(f"[OK] CSE Stock: {str(parsed)[:200]}...")
else:
print("[WARN] scrape_cse_stock_data not found in ToolSet")
except Exception as e:
print(f"[FAIL] CSE Stock: {e}")
# Test 3: Government Gazette
print("\n[3] Testing scrape_government_gazette...")
try:
gazette_tool = tools.get("scrape_government_gazette")
if gazette_tool:
result = gazette_tool.invoke({"keywords": None, "max_items": 3})
parsed = json.loads(result) if isinstance(result, str) else result
if "error" in parsed:
print(f"[WARN] Gazette returned error: {parsed['error']}")
else:
count = len(parsed.get('results', []))
print(f"[OK] Gazette: {count} items fetched")
else:
print("[WARN] scrape_government_gazette not found in ToolSet")
except Exception as e:
print(f"[FAIL] Gazette: {e}")
# Test 4: Parliament Minutes
print("\n[4] Testing scrape_parliament_minutes...")
try:
parliament_tool = tools.get("scrape_parliament_minutes")
if parliament_tool:
result = parliament_tool.invoke({"keywords": None, "max_items": 3})
parsed = json.loads(result) if isinstance(result, str) else result
if "error" in parsed:
print(f"[WARN] Parliament returned error: {parsed['error']}")
else:
count = len(parsed.get('results', []))
print(f"[OK] Parliament: {count} items fetched")
else:
print("[WARN] scrape_parliament_minutes not found in ToolSet")
except Exception as e:
print(f"[FAIL] Parliament: {e}")
# Test 5: Reddit (no auth needed)
print("\n[5] Testing scrape_reddit...")
try:
reddit_tool = tools.get("scrape_reddit")
if reddit_tool:
result = reddit_tool.invoke({"keywords": ["sri lanka"], "limit": 3})
parsed = json.loads(result) if isinstance(result, str) else result
if "error" in parsed:
print(f"[WARN] Reddit returned error: {parsed['error']}")
else:
count = len(parsed.get('results', parsed.get('posts', [])))
print(f"[OK] Reddit: {count} posts fetched")
else:
print("[WARN] scrape_reddit not found in ToolSet")
except Exception as e:
print(f"[FAIL] Reddit: {e}")
print("\n" + "="*60)
print("[DONE] News & official source tools test complete")
print("="*60)
|